Iceberg support partitioning on a nested field

[Buktoria]

Problem

When querying an iceberg table which is partitioned, you can get the following error.
java.lang.IllegalArgumentException: columns is empty.

java.lang.IllegalArgumentException: columns is empty
	at io.trino.spi.connector.DiscretePredicates.<init>(DiscretePredicates.java:31)
	at io.trino.plugin.iceberg.IcebergMetadata.getTableProperties(IcebergMetadata.java:368)
	at io.trino.plugin.base.classloader.ClassLoaderSafeConnectorMetadata.getTableProperties(ClassLoaderSafeConnectorMetadata.java:764)
	at io.trino.metadata.MetadataManager.getTableProperties(MetadataManager.java:489)
	at io.trino.sql.planner.iterative.rule.DetermineTableScanNodePartitioning.apply(DetermineTableScanNodePartitioning.java:59)
	at io.trino.sql.planner.iterative.rule.DetermineTableScanNodePartitioning.apply(DetermineTableScanNodePartitioning.java:33)

...

Deep Dive into the problem

This columns value is returning an empty list

  // Extract identity partition columns
  Map<Integer, IcebergColumnHandle> columns = getColumns(icebergTable.schema(), typeManager).stream()
          .filter(column -> partitionSourceIds.contains(column.getId()))
          .collect(toImmutableMap(IcebergColumnHandle::getId, Function.identity()));

If we take a look at the getColumns we can see that this is a simple iteration over the return value of the schema.columns() from the iceberg api. The problem is that some of these fields can be nested. This method is not unpacking those columns.

public static List<IcebergColumnHandle> getColumns(Schema schema, TypeManager typeManager)
{
    return schema.columns().stream()
            .map(column -> IcebergColumnHandle.create(column, typeManager))
            .collect(toImmutableList());
}

Now go back to to see how that original columns variable is being generated we see a filter being applied, to the top level column that we got from getColumns.

  // Extract identity partition columns
  Map<Integer, IcebergColumnHandle> columns = getColumns(icebergTable.schema(), typeManager).stream()
          .filter(column -> partitionSourceIds.contains(column.getId()))
          .collect(toImmutableMap(IcebergColumnHandle::getId, Function.identity()));

It means that all columns get filtered out because none of those columns are the partition field, hence getting the columns is empty when building

discretePredicates = new DiscretePredicates(
        columns.values().stream()
                .map(ColumnHandle.class::cast)
                .collect(toImmutableList()),
        discreteTupleDomain);

and then error thrown here

  public DiscretePredicates(List<ColumnHandle> columns, Iterable<TupleDomain<ColumnHandle>> predicates)
  {
      requireNonNull(columns, "columns is null");
      if (columns.isEmpty()) {
          throw new IllegalArgumentException("columns is empty");

Solution

This pr does three things main things.

• When getting table column handles from IcebergMetadata we get all column, including the ones that are nested.
• When looking for Iceberg column partitions we dig out the partition column and store the index positions to get that column from the schema (sourceIds).
• The Iceberg page partitioner also digs out the partition column block from a page with the stored sourceIds.

Closes: #5458

[findepi]

use fixed time in tests to ensure the test is reproducible

[findepi]

please inline createTable and insertSql -- they are used once

(same below)

[findepi]

assert on the result values, not just count

assert(query("...."))
  .matches("SELECT val1, val2")

[findepi]

i would assume that in this test the partition field is all that matters; do we need to have all these data columns?

[findepi]

how is second table different from the first?
ie what is the important difference?

[findepi]

thanks for the test. please also add one in TestIcebergSparkCompatibility to make sure our nested-field-partitioned writes are correctly readable by spark

we should also test the reverse.

note that TestIcebergSparkCompatibility is a product test, so to run it you need

./mvnw clean install -DskipTests # project needs to be built first
bin/ptl test run --environment singlenode-spark-iceberg -- -t TestIcebergSparkCompatibility

[Buktoria]

what is this bin/ptl command? I this something that should be created by that ./mvnw clean install -DskipTests command? There is no bin directory created after running the instal of the project on my side locally.

[alexjo2144]

It was moved into the testing directory ./testing/bin/ptl

[alexjo2144]

It'd be good to have a query with a WHERE clause in this test as well, to verify that the predicate pushdown on the nested column works properly. Pushdown on partition columns is guaranteed by the connector so the engine won't repeat the filter.

[findepi]

This should be more strict. For example, shouldn't match a...b.

IDENTIFIER = "[a-z_][a-z0-9_]*";
NAME = IDENTIFIER + "(\\." + NAME + ")*";

also, since we thus allow partitioning on a function from a field (like year(rec.timestamp)), we should add some test coverage for such usage too

[findepi]

in #9354 @djsstarburst argues that using getChildren is incorrect.

@djsstarburst would you want to comment here?

[findepi]

same as above

[findepi]

@Buktoria than you for your PR!
added couple initial comments.

cc @losipiuk @alexjo2144

[mosabua]

👋 @Buktoria - this PR has become inactive. If you're still interested in working on it, please let us know, and we can try to get reviewers to help with that.

We're working on closing out old and inactive PRs, so if you're too busy or this has too many merge conflicts to be worth picking back up, we'll be making another pass to close it out in a few weeks.

[Buktoria]

Closing this, its too far behind to be able to merge.