Skip to content

Fix Iceberg statistics for empty table #12316

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
findepi merged 5 commits into from
May 11, 2022
Merged

Fix Iceberg statistics for empty table #12316

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ce643a7
Fix procedure handle toString
findepi May 10, 2022
6cd9db6
Make blockToNativeValue public
findepi May 9, 2022
261ac26
Use query-based assertions instead of computeActual+MaterializedResult
findepi May 10, 2022
02f55b3
Inline redundant variable
findepi May 10, 2022
7381fa2
Fix Iceberg statistics for empty table
findepi May 10, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions core/trino-main/src/main/java/io/trino/operator/DynamicFilterSourceOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import static com.google.common.base.Verify.verify;
import static io.trino.operator.aggregation.TypedSet.createUnboundedEqualityTypedSet;
import static io.trino.spi.predicate.Range.range;
import static io.trino.spi.predicate.Utils.blockToNativeValue;
import static io.trino.spi.type.DoubleType.DOUBLE;
import static io.trino.spi.type.RealType.REAL;
import static io.trino.spi.type.TypeUtils.isFloatingPointNaN;
Expand Down Expand Up @@ -377,8 +378,8 @@ public void finish()
domainsBuilder.put(channels.get(channelIndex).filterId, Domain.none(type));
continue;
}
Object min = readNativeValue(type, minValues[channelIndex], 0);
Object max = readNativeValue(type, maxValues[channelIndex], 0);
Object min = blockToNativeValue(type, minValues[channelIndex]);
Object max = blockToNativeValue(type, maxValues[channelIndex]);
Domain domain = Domain.create(
ValueSet.ofRanges(range(type, min, true, max, true)),
false);
Expand Down
2 changes: 1 addition & 1 deletion core/trino-spi/src/main/java/io/trino/spi/predicate/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ public static Block nativeValueToBlock(Type type, @Nullable Object object)
return blockBuilder.build();
}

static Object blockToNativeValue(Type type, Block block)
public static Object blockToNativeValue(Type type, Block block)
{
if (block.getPositionCount() != 1) {
throw new IllegalArgumentException("Block should have exactly one position, but has: " + block.getPositionCount());
Expand Down
6 changes: 2 additions & 4 deletions plugin/trino-base-jdbc/src/main/java/io/trino/plugin/jdbc/QueryParameter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,9 +22,8 @@
import java.util.Objects;
import java.util.Optional;

import static com.google.common.base.Preconditions.checkArgument;
import static io.trino.spi.predicate.Utils.blockToNativeValue;
import static io.trino.spi.predicate.Utils.nativeValueToBlock;
import static io.trino.spi.type.TypeUtils.readNativeValue;
import static java.util.Objects.requireNonNull;

public final class QueryParameter
Expand All @@ -45,8 +44,7 @@ public static QueryParameter fromValueAsBlock(JdbcTypeHandle jdbcType, Type type
{
requireNonNull(type, "type is null");
requireNonNull(valueBlock, "valueBlock is null");
checkArgument(valueBlock.getPositionCount() == 1, "The block should have exactly one position, got %s", valueBlock.getPositionCount());
Optional<Object> value = Optional.ofNullable(readNativeValue(type, valueBlock, 0));
Optional<Object> value = Optional.ofNullable(blockToNativeValue(type, valueBlock));
return new QueryParameter(jdbcType, type, value);
}

Expand Down
3 changes: 2 additions & 1 deletion plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,7 @@
import static io.trino.spi.predicate.Range.lessThanOrEqual;
import static io.trino.spi.predicate.Range.range;
import static io.trino.spi.predicate.TupleDomain.withColumnDomains;
import static io.trino.spi.predicate.Utils.blockToNativeValue;
import static io.trino.spi.predicate.ValueSet.ofRanges;
import static io.trino.spi.statistics.ColumnStatisticType.MAX_VALUE;
import static io.trino.spi.statistics.ColumnStatisticType.NUMBER_OF_DISTINCT_VALUES_SUMMARY;
Expand Down Expand Up @@ -2047,7 +2048,7 @@ private static Map<String, DeltaLakeColumnStatistics> toDeltaLakeColumnStatistic
return DeltaLakeColumnStatistics.create(HyperLogLog.newInstance(4096)); // empty HLL with number of buckets used by $approx_set
}
else {
Slice serializedSummary = HyperLogLogType.HYPER_LOG_LOG.getSlice(entry.getValue(), 0);
Slice serializedSummary = (Slice) blockToNativeValue(HyperLogLogType.HYPER_LOG_LOG, entry.getValue());
return DeltaLakeColumnStatistics.create(HyperLogLog.newInstance(serializedSummary));
}
}));
Expand Down
12 changes: 9 additions & 3 deletions plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/TableStatisticsMaker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,13 +63,17 @@ public static TableStatistics getTableStatistics(TypeManager typeManager, Iceber
private TableStatistics makeTableStatistics(IcebergTableHandle tableHandle)
{
if (tableHandle.getSnapshotId().isEmpty()) {
return TableStatistics.empty();
return TableStatistics.builder()
Copy link
Copy Markdown
Member

@losipiuk losipiuk May 10, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: a constant for TableStatistics.builder() .setRowCount(Estimate.of(0)) .build();?

Copy link
Copy Markdown
Member Author

@findepi findepi May 10, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i would call it "empty_table_statisics" but then it would look the same as TableStatistics.empty();
maybe we should rename TableStatistics.empty() in a follow-up?

Copy link
Copy Markdown
Member

@losipiuk losipiuk May 10, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah - it makes sense

.setRowCount(Estimate.of(0))
.build();
}

TupleDomain<IcebergColumnHandle> enforcedPredicate = tableHandle.getEnforcedPredicate();

if (enforcedPredicate.isNone()) {
return TableStatistics.empty();
return TableStatistics.builder()
.setRowCount(Estimate.of(0))
.build();
}

Schema icebergTableSchema = icebergTable.schema();
Expand Down Expand Up @@ -111,7 +115,9 @@ private TableStatistics makeTableStatistics(IcebergTableHandle tableHandle)
IcebergStatistics summary = icebergStatisticsBuilder.build();

if (summary.getFileCount() == 0) {
return TableStatistics.empty();
return TableStatistics.builder()
.setRowCount(Estimate.of(0))
.build();
}

ImmutableMap.Builder<ColumnHandle, ColumnStatistics> columnHandleBuilder = ImmutableMap.builder();
Expand Down
5 changes: 3 additions & 2 deletions ...eberg/src/main/java/io/trino/plugin/iceberg/procedure/IcebergDeleteOrphanFilesHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import com.fasterxml.jackson.annotation.JsonProperty;
import io.airlift.units.Duration;

import static com.google.common.base.MoreObjects.toStringHelper;
import static java.util.Objects.requireNonNull;

public class IcebergDeleteOrphanFilesHandle
Expand All @@ -39,8 +40,8 @@ public Duration getRetentionThreshold()
@Override
public String toString()
{
return new StringBuilder()
.append("retentionThreshold=").append(retentionThreshold)
return toStringHelper(this)
.add("retentionThreshold", retentionThreshold)
.toString();
}
}
5 changes: 3 additions & 2 deletions ...iceberg/src/main/java/io/trino/plugin/iceberg/procedure/IcebergExpireSnapshotsHandle.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import com.fasterxml.jackson.annotation.JsonProperty;
import io.airlift.units.Duration;

import static com.google.common.base.MoreObjects.toStringHelper;
import static java.util.Objects.requireNonNull;

public class IcebergExpireSnapshotsHandle
Expand All @@ -39,8 +40,8 @@ public Duration getRetentionThreshold()
@Override
public String toString()
{
return new StringBuilder()
.append("retentionThreshold:").append(retentionThreshold)
return toStringHelper(this)
.add("retentionThreshold", retentionThreshold)
.toString();
}
}
62 changes: 30 additions & 32 deletions plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1789,29 +1789,29 @@ public void testBasicTableStatistics()
{
String tableName = "test_basic_table_statistics";
assertUpdate(format("CREATE TABLE %s (col REAL)", tableName));
String insertStart = format("INSERT INTO %s", tableName);
assertUpdate(insertStart + " VALUES -10", 1);
assertUpdate(insertStart + " VALUES 100", 1);

// SHOW STATS returns rows of the form: column_name, data_size, distinct_values_count, nulls_fractions, row_count, low_value, high_value
assertThat(query("SHOW STATS FOR " + tableName))
.skippingTypesCheck()
.matches("VALUES " +
" ('col', 0e0, 0e0, 1e0, NULL, NULL, NULL), " +
" (NULL, NULL, NULL, NULL, 0e0, NULL, NULL)");

MaterializedResult result = computeActual("SHOW STATS FOR " + tableName);
MaterializedResult expectedStatistics =
resultBuilder(getSession(), VARCHAR, DOUBLE, DOUBLE, DOUBLE, DOUBLE, VARCHAR, VARCHAR)
.row("col", null, null, 0.0, null, "-10.0", "100.0")
.row(null, null, null, null, 2.0, null, null)
.build();
assertEquals(result, expectedStatistics);
assertUpdate("INSERT INTO " + tableName + " VALUES -10", 1);
assertUpdate("INSERT INTO " + tableName + " VALUES 100", 1);

assertUpdate(insertStart + " VALUES 200", 1);
assertThat(query("SHOW STATS FOR " + tableName))
.skippingTypesCheck()
.matches("VALUES " +
" ('col', NULL, NULL, 0e0, NULL, '-10.0', '100.0'), " +
" (NULL, NULL, NULL, NULL, 2e0, NULL, NULL)");

result = computeActual("SHOW STATS FOR " + tableName);
expectedStatistics =
resultBuilder(getSession(), VARCHAR, DOUBLE, DOUBLE, DOUBLE, DOUBLE, VARCHAR, VARCHAR)
.row("col", null, null, 0.0, null, "-10.0", "200.0")
.row(null, null, null, null, 3.0, null, null)
.build();
assertEquals(result, expectedStatistics);
assertUpdate("INSERT INTO " + tableName + " VALUES 200", 1);

assertThat(query("SHOW STATS FOR " + tableName))
.skippingTypesCheck()
.matches("VALUES " +
" ('col', NULL, NULL, 0e0, NULL, '-10.0', '200.0'), " +
" (NULL, NULL, NULL, NULL, 3e0, NULL, NULL)");

dropTable(tableName);
}
Expand All @@ -1821,9 +1821,8 @@ public void testMultipleColumnTableStatistics()
{
String tableName = "test_multiple_table_statistics";
assertUpdate(format("CREATE TABLE %s (col1 REAL, col2 INTEGER, col3 DATE)", tableName));
String insertStart = format("INSERT INTO %s", tableName);
assertUpdate(insertStart + " VALUES (-10, -1, DATE '2019-06-28')", 1);
assertUpdate(insertStart + " VALUES (100, 10, DATE '2020-01-01')", 1);
assertUpdate("INSERT INTO " + tableName + " VALUES (-10, -1, DATE '2019-06-28')", 1);
assertUpdate("INSERT INTO " + tableName + " VALUES (100, 10, DATE '2020-01-01')", 1);

MaterializedResult result = computeActual("SHOW STATS FOR " + tableName);

Expand All @@ -1836,7 +1835,7 @@ public void testMultipleColumnTableStatistics()
.build();
assertEquals(result, expectedStatistics);

assertUpdate(insertStart + " VALUES (200, 20, DATE '2020-06-28')", 1);
assertUpdate("INSERT INTO " + tableName + " VALUES (200, 20, DATE '2020-06-28')", 1);
result = computeActual("SHOW STATS FOR " + tableName);
expectedStatistics =
resultBuilder(getSession(), VARCHAR, DOUBLE, DOUBLE, DOUBLE, DOUBLE, VARCHAR, VARCHAR)
Expand All @@ -1847,11 +1846,11 @@ public void testMultipleColumnTableStatistics()
.build();
assertEquals(result, expectedStatistics);

assertUpdate(insertStart + " VALUES " + IntStream.rangeClosed(21, 25)
assertUpdate("INSERT INTO " + tableName + " VALUES " + IntStream.rangeClosed(21, 25)
.mapToObj(i -> format("(200, %d, DATE '2020-07-%d')", i, i))
.collect(joining(", ")), 5);

assertUpdate(insertStart + " VALUES " + IntStream.rangeClosed(26, 30)
assertUpdate("INSERT INTO " + tableName + " VALUES " + IntStream.rangeClosed(26, 30)
.mapToObj(i -> format("(NULL, %d, DATE '2020-06-%d')", i, i))
.collect(joining(", ")), 5);

Expand All @@ -1874,9 +1873,8 @@ public void testPartitionedTableStatistics()
{
assertUpdate("CREATE TABLE iceberg.tpch.test_partitioned_table_statistics (col1 REAL, col2 BIGINT) WITH (partitioning = ARRAY['col2'])");

String insertStart = "INSERT INTO test_partitioned_table_statistics";
assertUpdate(insertStart + " VALUES (-10, -1)", 1);
assertUpdate(insertStart + " VALUES (100, 10)", 1);
assertUpdate("INSERT INTO test_partitioned_table_statistics VALUES (-10, -1)", 1);
assertUpdate("INSERT INTO test_partitioned_table_statistics VALUES (100, 10)", 1);

MaterializedResult result = computeActual("SHOW STATS FOR iceberg.tpch.test_partitioned_table_statistics");
assertEquals(result.getRowCount(), 3);
Expand All @@ -1896,11 +1894,11 @@ public void testPartitionedTableStatistics()
MaterializedRow row2 = result.getMaterializedRows().get(2);
assertEquals(row2.getField(4), 2.0);

assertUpdate(insertStart + " VALUES " + IntStream.rangeClosed(1, 5)
assertUpdate("INSERT INTO test_partitioned_table_statistics VALUES " + IntStream.rangeClosed(1, 5)
.mapToObj(i -> format("(%d, 10)", i + 100))
.collect(joining(", ")), 5);

assertUpdate(insertStart + " VALUES " + IntStream.rangeClosed(6, 10)
assertUpdate("INSERT INTO test_partitioned_table_statistics VALUES " + IntStream.rangeClosed(6, 10)
.mapToObj(i -> "(NULL, 10)")
.collect(joining(", ")), 5);

Expand All @@ -1921,7 +1919,7 @@ public void testPartitionedTableStatistics()
row2 = result.getMaterializedRows().get(2);
assertEquals(row2.getField(4), 12.0);

assertUpdate(insertStart + " VALUES " + IntStream.rangeClosed(6, 10)
assertUpdate("INSERT INTO test_partitioned_table_statistics VALUES " + IntStream.rangeClosed(6, 10)
.mapToObj(i -> "(100, NULL)")
.collect(joining(", ")), 5);

Expand Down Expand Up @@ -3295,7 +3293,7 @@ public void testExplainExpireSnapshotOutput()
assertUpdate("INSERT INTO " + tableName + " VALUES ('two', 2)", 1);

assertExplain("EXPLAIN ALTER TABLE " + tableName + " EXECUTE EXPIRE_SNAPSHOTS (retention_threshold => '0s')",
"SimpleTableExecute\\[iceberg:schemaTableName:tpch.test_expiring_snapshots.*\\{retentionThreshold:0\\.00s}.*");
"SimpleTableExecute\\[iceberg:schemaTableName:tpch.test_expiring_snapshots.*\\{retentionThreshold=0\\.00s}.*");
}

@Test
Expand Down