diff --git a/api/src/main/java/org/apache/iceberg/expressions/And.java b/api/src/main/java/org/apache/iceberg/expressions/And.java index 945a1c2536..bc006475cd 100644 --- a/api/src/main/java/org/apache/iceberg/expressions/And.java +++ b/api/src/main/java/org/apache/iceberg/expressions/And.java @@ -19,7 +19,10 @@ package org.apache.iceberg.expressions; -public class And implements Expression { +import org.apache.iceberg.StructLike; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +public class And implements Expression, Bound { private final Expression left; private final Expression right; @@ -47,6 +50,21 @@ public Expression negate() { return Expressions.or(left.negate(), right.negate()); } + @Override + public BoundReference ref() { + return null; + } + + @Override + public Boolean eval(StructLike struct) { + Preconditions.checkNotNull(left, "Left expression cannot be null."); + Preconditions.checkNotNull(right, "Right expression cannot be null."); + if (!(left instanceof Bound) || !(right instanceof Bound)) { + throw new IllegalStateException("Unbound predicate not expected"); + } + return ((Bound) left).eval(struct) && ((Bound) right).eval(struct); + } + @Override public String toString() { return String.format("(%s and %s)", left, right); diff --git a/api/src/main/java/org/apache/iceberg/expressions/Or.java b/api/src/main/java/org/apache/iceberg/expressions/Or.java index b41ef7f676..f98f2bf573 100644 --- a/api/src/main/java/org/apache/iceberg/expressions/Or.java +++ b/api/src/main/java/org/apache/iceberg/expressions/Or.java @@ -19,7 +19,10 @@ package org.apache.iceberg.expressions; -public class Or implements Expression { +import org.apache.iceberg.StructLike; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +public class Or implements Expression, Bound { private final Expression left; private final Expression right; @@ -47,6 +50,21 @@ public Expression negate() { return Expressions.and(left.negate(), right.negate()); } + @Override + public BoundReference ref() { + return null; + } + + @Override + public Boolean eval(StructLike struct) { + Preconditions.checkNotNull(left, "Left expression cannot be null."); + Preconditions.checkNotNull(right, "Right expression cannot be null."); + if (!(left instanceof Bound) || !(right instanceof Bound)) { + throw new IllegalStateException("Unbound predicate not expected"); + } + return ((Bound) left).eval(struct) || ((Bound) right).eval(struct); + } + @Override public String toString() { return String.format("(%s or %s)", left, right); diff --git a/api/src/main/java/org/apache/iceberg/types/Conversions.java b/api/src/main/java/org/apache/iceberg/types/Conversions.java index d6ef24449b..421333f9f5 100644 --- a/api/src/main/java/org/apache/iceberg/types/Conversions.java +++ b/api/src/main/java/org/apache/iceberg/types/Conversions.java @@ -71,6 +71,8 @@ public static Object fromPartitionString(Type type, String asString) { return new BigDecimal(asString); case DATE: return Literal.of(asString).to(Types.DateType.get()).value(); + case TIMESTAMP: + return Literal.of(asString).to(Types.TimestampType.withoutZone()).value(); default: throw new UnsupportedOperationException( "Unsupported type for fromPartitionString: " + type); diff --git a/hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/HiveExpressions.java b/hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/HiveExpressions.java index 1363d77ce0..3ce14f37fc 100644 --- a/hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/HiveExpressions.java +++ b/hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/HiveExpressions.java @@ -19,6 +19,10 @@ package org.apache.iceberg.hive.legacy; +import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.time.temporal.ChronoUnit; import java.util.List; import java.util.Set; import java.util.stream.Collectors; @@ -29,6 +33,7 @@ import org.apache.iceberg.expressions.Literal; import org.apache.iceberg.expressions.UnboundPredicate; import org.apache.iceberg.expressions.UnboundTerm; +import org.apache.iceberg.types.Type; class HiveExpressions { @@ -238,6 +243,7 @@ public Expression predicate(UnboundPredicate pred) { } private static class ExpressionToPartitionFilterString extends ExpressionVisitors.ExpressionVisitor { + private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC); private static final ExpressionToPartitionFilterString INSTANCE = new ExpressionToPartitionFilterString(); private ExpressionToPartitionFilterString() { @@ -274,11 +280,6 @@ public String or(String leftResult, String rightResult) { @Override public String predicate(BoundPredicate pred) { - throw new IllegalStateException("Bound predicate not expected: " + pred.getClass().getName()); - } - - @Override - public String predicate(UnboundPredicate pred) { switch (pred.op()) { case LT: case LT_EQ: @@ -286,14 +287,22 @@ public String predicate(UnboundPredicate pred) { case GT_EQ: case EQ: case NOT_EQ: - return getBinaryExpressionString(pred.ref().name(), pred.op(), pred.literal()); + return getBinaryExpressionString(pred); default: throw new IllegalStateException("Unexpected operator in Hive partition filter string: " + pred.op()); } } - private String getBinaryExpressionString(String columnName, Expression.Operation op, Literal lit) { - return String.format("( %s %s %s )", columnName, getOperationString(op), getLiteralValue(lit)); + @Override + public String predicate(UnboundPredicate pred) { + throw new IllegalStateException("Unbound predicate not expected: " + pred.getClass().getName()); + } + + private String getBinaryExpressionString(BoundPredicate pred) { + String columnName = pred.ref().field().name(); + String opName = getOperationString(pred.op()); + String litValue = getLiteralValue(pred.asLiteralPredicate().literal(), pred.ref().type()); + return String.format("( %s %s %s )", columnName, opName, litValue); } private String getOperationString(Expression.Operation op) { @@ -315,8 +324,11 @@ private String getOperationString(Expression.Operation op) { } } - private String getLiteralValue(Literal lit) { + private String getLiteralValue(Literal lit, Type type) { Object value = lit.value(); + if (type.typeId() == Type.TypeID.DATE) { + value = EPOCH.plus((Integer) value, ChronoUnit.DAYS).toLocalDate().toString(); + } if (value instanceof String) { String escapedString = ((String) value).replace("'", "\\'"); return String.format("'%s'", escapedString); diff --git a/hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/LegacyHiveTableOperations.java b/hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/LegacyHiveTableOperations.java index 1620d67cf0..dbd1773114 100644 --- a/hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/LegacyHiveTableOperations.java +++ b/hive-metastore/src/main/java/org/apache/iceberg/hive/legacy/LegacyHiveTableOperations.java @@ -19,12 +19,19 @@ package org.apache.iceberg.hive.legacy; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.ZoneOffset; +import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeFormatterBuilder; +import java.time.temporal.ChronoField; import java.util.List; import java.util.Map; import java.util.Set; import java.util.stream.Collectors; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.hive.metastore.api.MetaException; import org.apache.hadoop.hive.metastore.api.Partition; import org.apache.iceberg.BaseMetastoreTableOperations; import org.apache.iceberg.DataFile; @@ -36,6 +43,9 @@ import org.apache.iceberg.StructLike; import org.apache.iceberg.TableMetadata; import org.apache.iceberg.TableProperties; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.expressions.Binder; +import org.apache.iceberg.expressions.Bound; import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.hadoop.HadoopFileIO; @@ -48,6 +58,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.types.Types; import org.apache.thrift.TException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -109,7 +120,7 @@ protected void doRefresh() { /** * Returns an {@link Iterable} of {@link Iterable}s of {@link DataFile}s which belong to the current table and * match the partition predicates from the given expression. - * + *

* Each element in the outer {@link Iterable} maps to an {@link Iterable} of {@link DataFile}s originating from the * same directory */ @@ -170,9 +181,11 @@ private List getDirectoryInfosByFilter(Expression expression) { .map(id -> current().schema().findColumnName(id)) .collect(Collectors.toSet()); Expression simplified = HiveExpressions.simplifyPartitionFilter(expression, partitionColumnNames); + Types.StructType partitionSchema = current().spec().partitionType(); LOG.info("Simplified expression for {}.{} to {}", databaseName, tableName, simplified); - final List partitions; + List partitions; + Expression boundExpression; if (simplified.equals(Expressions.alwaysFalse())) { // If simplifyPartitionFilter returns FALSE, no partitions are going to match the filter expression partitions = ImmutableList.of(); @@ -181,10 +194,47 @@ private List getDirectoryInfosByFilter(Expression expression) { partitions = metaClients.run(client -> client.listPartitionsByFilter( databaseName, tableName, null, (short) -1)); } else { - String partitionFilterString = HiveExpressions.toPartitionFilterString(simplified); + boundExpression = Binder.bind(partitionSchema, simplified, false); + String partitionFilterString = HiveExpressions.toPartitionFilterString(boundExpression); LOG.info("Listing partitions for {}.{} with filter string: {}", databaseName, tableName, partitionFilterString); - partitions = metaClients.run( - client -> client.listPartitionsByFilter(databaseName, tableName, partitionFilterString, (short) -1)); + try { + // We first try to use HMS API call to get the filtered partitions. + partitions = metaClients.run( + client -> client.listPartitionsByFilter(databaseName, tableName, partitionFilterString, (short) -1)); + } catch (MetaException e) { + // If the above HMS call fails, we here try to do the partition filtering ourselves, + // by evaluating all the partitions we got back from HMS against the boundExpression, + // if the evaluation results in true, we include such partition, if false, we filter. + List allPartitions = metaClients.run( + client -> client.listPartitionsByFilter(databaseName, tableName, null, (short) -1)); + partitions = allPartitions.stream().filter(partition -> { + GenericRecord record = GenericRecord.create(partitionSchema); + for (int i = 0; i < record.size(); i++) { + String value = partition.getValues().get(i); + switch (partitionSchema.fields().get(i).type().typeId()) { + case DATE: + record.set(i, + (int) LocalDate.parse(value).toEpochDay()); + break; + case TIMESTAMP: + // This format seems to be matching the hive timestamp column partition string literal value + record.set(i, + LocalDateTime.parse(value, + new DateTimeFormatterBuilder() + .parseLenient() + .append(DateTimeFormatter.ISO_LOCAL_DATE_TIME) + .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true) + .toFormatter()) + .toInstant(ZoneOffset.UTC).toEpochMilli() * 1000); + break; + default: + record.set(i, partition.getValues().get(i)); + break; + } + } + return ((Bound) boundExpression).eval(record); + }).collect(Collectors.toList()); + } } return LegacyHiveTableUtils.toDirectoryInfos(partitions, current().spec()); @@ -199,7 +249,7 @@ private List getDirectoryInfosByFilter(Expression expression) { } private static DataFile createDataFile(FileStatus fileStatus, PartitionSpec partitionSpec, StructLike partitionData, - FileFormat format) { + FileFormat format) { DataFiles.Builder builder = DataFiles.builder(partitionSpec) .withPath(fileStatus.getPath().toString()) .withFormat(format) diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestHiveExpressions.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestHiveExpressions.java index 2d2e8b40eb..9acfc3d41d 100644 --- a/hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestHiveExpressions.java +++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestHiveExpressions.java @@ -36,7 +36,6 @@ import static org.apache.iceberg.expressions.Expressions.notNull; import static org.apache.iceberg.expressions.Expressions.or; import static org.apache.iceberg.hive.legacy.HiveExpressions.simplifyPartitionFilter; -import static org.apache.iceberg.hive.legacy.HiveExpressions.toPartitionFilterString; public class TestHiveExpressions { @@ -118,10 +117,4 @@ public void testSimplifyRemoveNonPartitionColumnsWithinNot2() { Expression expected = alwaysTrue(); Assert.assertEquals(expected.toString(), simplifyPartitionFilter(input, ImmutableSet.of("pcol")).toString()); } - - @Test - public void testToPartitionFilterStringEscapeStringLiterals() { - Expression input = equal("pcol", "s'1"); - Assert.assertEquals("( pcol = 's\\'1' )", toPartitionFilterString(input)); - } } diff --git a/hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestLegacyHiveTableScan.java b/hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestLegacyHiveTableScan.java index 55667089a8..9baa05344f 100644 --- a/hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestLegacyHiveTableScan.java +++ b/hive-metastore/src/test/java/org/apache/iceberg/hive/legacy/TestLegacyHiveTableScan.java @@ -25,6 +25,12 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.time.Instant; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.time.temporal.ChronoUnit; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -38,7 +44,6 @@ import org.apache.hadoop.hive.metastore.api.StorageDescriptor; import org.apache.hadoop.hive.metastore.api.Table; import org.apache.hadoop.hive.serde2.avro.AvroSerdeUtils; -import org.apache.iceberg.AssertHelpers; import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileScanTask; import org.apache.iceberg.avro.AvroSchemaUtil; @@ -62,13 +67,32 @@ public class TestLegacyHiveTableScan extends HiveMetastoreTest { + private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC); + private static final List DATA_COLUMNS = ImmutableList.of( new FieldSchema("strCol", "string", ""), new FieldSchema("intCol", "int", "")); + private static final List PARTITION_COLUMNS = ImmutableList.of( new FieldSchema("pcol", "string", ""), new FieldSchema("pIntCol", "int", "")); + private static final List PARTITION_COLUMNS_2 = ImmutableList.of( + new FieldSchema("pcol", "string", ""), + new FieldSchema("pIntCol", "int", ""), + new FieldSchema("pDateCol", "date", "")); + + private static final List PARTITION_COLUMNS_3 = ImmutableList.of( + new FieldSchema("pcol", "string", ""), + new FieldSchema("pTsCol", "timestamp", "")); + + private static final List PARTITION_COLUMNS_4 = ImmutableList.of( + new FieldSchema("pStringCol", "string", ""), + new FieldSchema("pIntCol", "int", ""), + new FieldSchema("pCharCol", "char(1)", ""), + new FieldSchema("pVarcharCol", "varchar(10)", ""), + new FieldSchema("pDateCol", "date", "")); + private static HiveCatalog legacyCatalog; private static Path dbPath; @@ -140,14 +164,68 @@ public void testHiveScanMultiPartitionWithFilter() throws Exception { hiveScan(table, Expressions.equal("pcol", "ds"))); } + @Test + public void testHiveScanMultiPartitionWithFilterDate() throws Exception { + String tableName = "multi_partition_with_filter_date"; + Table table = createTable(tableName, DATA_COLUMNS, PARTITION_COLUMNS_2); + addPartition(table, ImmutableList.of("ds", 1, LocalDate.of(2019, 4, 14)), AVRO, "A"); + addPartition(table, ImmutableList.of("ds", 1, LocalDate.of(2021, 6, 2)), AVRO, "B"); + // 18000 is the # of days since epoch for 2019-04-14, + // this representation matches how Iceberg internally store the value in DateLiteral. + filesMatch( + ImmutableMap.of("pcol=ds/pIntCol=1/pDateCol=2019-04-14/A", AVRO), + hiveScan(table, Expressions.equal("pDateCol", 18000))); + } + + @Test + public void testHiveScanMultiPartitionWithFilterTs() throws Exception { + LocalDateTime ldt = EPOCH.plus(1000000000111000L, ChronoUnit.MICROS).toLocalDateTime(); + + String tableName = "multi_partition_with_filter_ts"; + Table table = createTable(tableName, DATA_COLUMNS, PARTITION_COLUMNS_3); + addPartition(table, ImmutableList.of("foo", ldt), AVRO, "A"); + addPartition(table, ImmutableList.of("bar", ldt), AVRO, "B"); + // 1000000000111000L microseconds since epoch correspond to 2001-09-09T01:46:40.111, + // this representation matches how Iceberg internally store the value in TimeStampLiteral. + filesMatch( + ImmutableMap.of("pcol=foo/pTsCol=2001-09-09T01:46:40.111/A", AVRO), + hiveScan(table, Expressions.and( + Expressions.equal("pCol", "foo"), Expressions.equal("pTsCol", 1000000000111000L)))); + } + @Test public void testHiveScanNonStringPartitionQuery() throws Exception { String tableName = "multi_partition_with_filter_on_non_string_partition_cols"; Table table = createTable(tableName, DATA_COLUMNS, PARTITION_COLUMNS); - AssertHelpers.assertThrows( - "Filtering on non string partition is not supported by ORM layer and we can enable direct sql only on mysql", - RuntimeException.class, "Failed to get partition info", - () -> hiveScan(table, Expressions.and(Expressions.equal("pcol", "ds"), Expressions.equal("pIntCol", "1")))); + filesMatch( + ImmutableMap.of(), + hiveScan(table, Expressions.and( + Expressions.equal("pcol", "ds"), Expressions.equal("pIntCol", 1)))); + } + + @Test + public void testHiveScanComplexNonStringPartitionQuery() throws Exception { + String tableName = "multi_partition_with_filter_on_complex_non_string_partition_cols"; + Table table = createTable(tableName, DATA_COLUMNS, PARTITION_COLUMNS_4); + addPartition(table, ImmutableList.of("foo", 0, "a", "xy", LocalDate.of(2019, 4, 14)), AVRO, "A"); + addPartition(table, ImmutableList.of("foo", 1, "a", "xy", LocalDate.of(2019, 4, 14)), AVRO, "B"); + addPartition(table, ImmutableList.of("foo", 1, "b", "xy", LocalDate.of(2019, 4, 14)), AVRO, "C"); + addPartition(table, ImmutableList.of("foo", 1, "b", "xyz", LocalDate.of(2019, 4, 14)), AVRO, "D"); + addPartition(table, ImmutableList.of("foo", 1, "b", "xyz", LocalDate.of(2020, 4, 14)), AVRO, "E"); + addPartition(table, ImmutableList.of("bar", 0, "a", "xy", LocalDate.of(2020, 4, 14)), AVRO, "F"); + + filesMatch( + ImmutableMap.of("pStringCol=bar/pIntCol=0/pCharCol=a/pVarcharCol=xy/pDateCol=2020-04-14/F", AVRO), + hiveScan(table, Expressions.equal("pstringcol", "bar"))); + filesMatch( + ImmutableMap.of("pStringCol=foo/pIntCol=1/pCharCol=b/pVarcharCol=xyz/pDateCol=2019-04-14/D", AVRO, + "pStringCol=foo/pIntCol=1/pCharCol=b/pVarcharCol=xyz/pDateCol=2020-04-14/E", AVRO), + hiveScan(table, Expressions.and(Expressions.equal("pcharcol", "b"), Expressions.equal("pvarcharcol", "xyz")))); + filesMatch( + ImmutableMap.of(), + hiveScan(table, Expressions.and( + Expressions.equal("pdatecol", "2020-04-14"), + Expressions.and(Expressions.equal("pcharcol", "b"), Expressions.equal("pvarcharcol", "xy"))))); } @Test @@ -290,7 +368,11 @@ private Map hiveScan(Table table, Expression filter) { return StreamSupport .stream(fileScanTasks.spliterator(), false) .collect(Collectors.toMap( - f -> tableLocation.relativize(Paths.get(URI.create(f.file().path().toString()))).toString().split("\\.")[0], + f -> { + String fullPath = tableLocation.relativize(Paths.get(URI.create(f.file().path().toString()))).toString(); + int idx = fullPath.lastIndexOf("."); + return fullPath.substring(0, idx); + }, f -> f.file().format())); }