Incremental pull should set the _hoodie_commit_time automatically from configs in HoodieInputFormat

[prazanna]

HoodieInputFormat should pick up the start_ts from the config and merge the _hoodie_commit_time predicate with the predicates set already. This allows the processing SQL to be exactly the same for backfill and incremental processing.

SET hoodie.<table_name>.consume.mode=LATEST
Select * from table1; // this would run against all the data in table1

SET hoodie.<table_name>.consume.mode=INCREMENTAL
SET hoodie.<table_name>.start.timestamp=20170101132319
Select * from table1; // this would run against only data inserted after 2017/01/01 12:23:19

Today we have to do

SET hoodie.<table_name>.consume.mode=INCREMENTAL
Select * from table1 where _hoodie_commit_ts > '20170101132319'; 

[vinothchandar]

This is the dead code sitting in HoodieInputFormat

/**
   * Clears out the filter expression (if this is not done, then ParquetReader will override the
   * FilterPredicate set)
   */
  private void clearOutExistingPredicate(JobConf job) {
    job.unset(TableScanDesc.FILTER_EXPR_CONF_STR);
  }

  /**
   * Constructs the predicate to push down to parquet storage. This creates the predicate for
   * `hoodie_commit_time` > 'start_commit_time' and ANDs with the existing predicate if one is
   * present already.
   */
  private FilterPredicate constructHoodiePredicate(JobConf job, String tableName, InputSplit split)
      throws IOException {
    FilterPredicate commitTimePushdown = constructCommitTimePushdownPredicate(job, tableName);
    LOG.info("Commit time predicate - " + commitTimePushdown.toString());
    FilterPredicate existingPushdown = constructHQLPushdownPredicate(job, split);
    LOG.info("Existing predicate - " + existingPushdown);

    FilterPredicate hoodiePredicate;
    if (existingPushdown != null) {
      hoodiePredicate = and(existingPushdown, commitTimePushdown);
    } else {
      hoodiePredicate = commitTimePushdown;
    }
    LOG.info("Hoodie Predicate - " + hoodiePredicate);
    return hoodiePredicate;
  }

  private FilterPredicate constructHQLPushdownPredicate(JobConf job, InputSplit split)
      throws IOException {
    String serializedPushdown = job.get(TableScanDesc.FILTER_EXPR_CONF_STR);
    String columnNamesString = job.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR);
    if (serializedPushdown == null || columnNamesString == null || serializedPushdown.isEmpty()
        || columnNamesString.isEmpty()) {
      return null;
    } else {
      SearchArgument sarg = SearchArgumentFactory
          .create(Utilities.deserializeExpression(serializedPushdown));
      final Path finalPath = ((FileSplit) split).getPath();
      final ParquetMetadata parquetMetadata = ParquetFileReader.readFooter(job, finalPath);
      final FileMetaData fileMetaData = parquetMetadata.getFileMetaData();
      return ParquetFilterPredicateConverter.toFilterPredicate(sarg, fileMetaData.getSchema());
    }
  }

  private FilterPredicate constructCommitTimePushdownPredicate(JobConf job, String tableName)
      throws IOException {
    String lastIncrementalTs = HoodieHiveUtil.readStartCommitTime(Job.getInstance(job), tableName);
    Operators.BinaryColumn sequenceColumn = binaryColumn(HoodieRecord.COMMIT_TIME_METADATA_FIELD);
    FilterPredicate p = gt(sequenceColumn, Binary.fromString(lastIncrementalTs));
    LOG.info("Setting predicate in InputFormat " + p.toString());
    return p;
  }

This also causes most of issues for Hive2 support, by bringing in some parquet-deps. Removing and simplifying life until we can fix this again.

[vinothchandar]

Moved to https://issues.apache.org/jira/browse/HUDI-59