Skip to content
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

Collect Delta extended statistics when creating table #15878

Merged
merged 7 commits into from
Feb 7, 2023
Merged

Collect Delta extended statistics when creating table #15878

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d7976ba
Fix typo
findepi Oct 11, 2022
0b55cb6
Remove unnecessary variable
findepi Oct 11, 2022
6b0b80b
Simplify map merging in Delta finishStatisticsCollection
findepi Oct 11, 2022
04165ef
Extract statistics update to separate method
pajaks Feb 2, 2023
2b0ae97
Collect Delta extended statistics during table creation
findepi Oct 11, 2022
22a27e6
empty
pajaks Feb 6, 2023
97e5fe8
empty
pajaks Feb 7, 2023
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
2 changes: 1 addition & 1 deletion core/trino-main/src/main/java/io/trino/sql/planner/StatisticsAggregationPlanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ public TableStatisticAggregation createStatisticsAggregation(TableStatisticsMeta
for (ColumnStatisticMetadata columnStatisticMetadata : statisticsMetadata.getColumnStatistics()) {
String columnName = columnStatisticMetadata.getColumnName();
Symbol inputSymbol = columnToSymbolMap.get(columnName);
verifyNotNull(inputSymbol, "inputSymbol is null");
verifyNotNull(inputSymbol, "no symbol for [%s] column, these columns exist: %s", columnName, columnToSymbolMap.keySet());
Type inputType = symbolAllocator.getTypes().get(inputSymbol);
verifyNotNull(inputType, "inputType is null for symbol: %s", inputSymbol);
ColumnStatisticsAggregation aggregation;
Expand Down
14 changes: 14 additions & 0 deletions plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@ public class DeltaLakeConfig
private Duration dynamicFilteringWaitTimeout = new Duration(0, SECONDS);
private boolean tableStatisticsEnabled = true;
private boolean extendedStatisticsEnabled = true;
private boolean collectExtendedStatisticsOnWrite = true;
private HiveCompressionCodec compressionCodec = HiveCompressionCodec.SNAPPY;
private long perTransactionMetastoreCacheMaximumSize = 1000;
private boolean deleteSchemaLocationsFallback;
Expand Down Expand Up @@ -346,6 +347,19 @@ public DeltaLakeConfig setExtendedStatisticsEnabled(boolean extendedStatisticsEn
return this;
}

public boolean isCollectExtendedStatisticsOnWrite()
{
return collectExtendedStatisticsOnWrite;
}

@Config("delta.extended-statistics.collect-on-write")
Copy link
Contributor

Choose a reason for hiding this comment

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

Do consider documenting this new property in delta-lake.rst - either in this PR or a follow-up PR

Copy link
Member Author

Choose a reason for hiding this comment

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

I would wait with documentation until other write operations are implemented if that's ok.

@ConfigDescription("Enables automatic column level extended statistics collection on write")
public DeltaLakeConfig setCollectExtendedStatisticsOnWrite(boolean collectExtendedStatisticsOnWrite)
{
this.collectExtendedStatisticsOnWrite = collectExtendedStatisticsOnWrite;
return this;
}

@NotNull
public HiveCompressionCodec getCompressionCodec()
{
Expand Down
147 changes: 105 additions & 42 deletions plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java
View file
Open in desktop

Large diffs are not rendered by default.

11 changes: 11 additions & 0 deletions .../trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeSessionProperties.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ public final class DeltaLakeSessionProperties
private static final String DYNAMIC_FILTERING_WAIT_TIMEOUT = "dynamic_filtering_wait_timeout";
private static final String TABLE_STATISTICS_ENABLED = "statistics_enabled";
public static final String EXTENDED_STATISTICS_ENABLED = "extended_statistics_enabled";
public static final String EXTENDED_STATISTICS_COLLECT_ON_WRITE = "extended_statistics_collect_on_write";
public static final String LEGACY_CREATE_TABLE_WITH_EXISTING_LOCATION_ENABLED = "legacy_create_table_with_existing_location_enabled";

private final List<PropertyMetadata<?>> sessionProperties;
Expand Down Expand Up @@ -163,6 +164,11 @@ public DeltaLakeSessionProperties(
"Enable using the CREATE TABLE statement to register an existing table",
deltaLakeConfig.isLegacyCreateTableWithExistingLocationEnabled(),
false),
booleanProperty(
EXTENDED_STATISTICS_COLLECT_ON_WRITE,
"Enables automatic column level extended statistics collection on write",
deltaLakeConfig.isCollectExtendedStatisticsOnWrite(),
false),
enumProperty(
COMPRESSION_CODEC,
"Compression codec to use when writing new data files",
Expand Down Expand Up @@ -258,6 +264,11 @@ public static boolean isLegacyCreateTableWithExistingLocationEnabled(ConnectorSe
return session.getProperty(LEGACY_CREATE_TABLE_WITH_EXISTING_LOCATION_ENABLED, Boolean.class);
}

public static boolean isCollectExtendedStatisticsColumnStatisticsOnWrite(ConnectorSession session)
{
return session.getProperty(EXTENDED_STATISTICS_COLLECT_ON_WRITE, Boolean.class);
}

public static HiveCompressionCodec getCompressionCodec(ConnectorSession session)
{
return session.getProperty(COMPRESSION_CODEC, HiveCompressionCodec.class);
Expand Down
37 changes: 18 additions & 19 deletions ...o-delta-lake/src/test/java/io/trino/plugin/deltalake/BaseDeltaLakeConnectorSmokeTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@
import static io.trino.SystemSessionProperties.ENABLE_DYNAMIC_FILTERING;
import static io.trino.SystemSessionProperties.JOIN_DISTRIBUTION_TYPE;
import static io.trino.plugin.deltalake.DeltaLakeQueryRunner.DELTA_CATALOG;
import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.EXTENDED_STATISTICS_COLLECT_ON_WRITE;
import static io.trino.plugin.deltalake.transactionlog.TransactionLogUtil.TRANSACTION_LOG_DIRECTORY;
import static io.trino.plugin.hive.TestingThriftHiveMetastoreBuilder.testingThriftHiveMetastoreBuilder;
import static io.trino.testing.TestingAccessControlManager.TestingPrivilegeType.DELETE_TABLE;
Expand Down Expand Up @@ -513,26 +514,21 @@ tableName, getLocationForTable(bucketName, tableName)),
@Test
public void testCreateTableAsStatistics()
{
String tableName = "test_ctats_stats_" + randomNameSuffix();
assertUpdate("CREATE TABLE " + tableName
+ " WITH ("
+ "location = '" + getLocationForTable(bucketName, tableName) + "'"
+ ")"
+ " AS SELECT * FROM tpch.sf1.nation", 25);

assertQuery(
"SHOW STATS FOR lineitem",
"SHOW STATS FOR " + tableName,
"VALUES " +
"('orderkey', NULL, NULL, 0.0, NULL, '1', '60000')," +
"('partkey', NULL, NULL, 0.0, NULL, '1', '2000')," +
"('suppkey', NULL, NULL, 0.0, NULL, '1', '100')," +
"('linenumber', NULL, NULL, 0.0, NULL, '1', '7')," +
"('quantity', NULL, NULL, 0.0, NULL, '1.0', '50.0')," +
"('extendedprice', NULL, NULL, 0.0, NULL, '904.0', '94949.5')," +
"('discount', NULL, NULL, 0.0, NULL, '0.0', '0.1')," +
"('tax', NULL, NULL, 0.0, NULL, '0.0', '0.08')," +
"('returnflag', NULL, NULL, 0.0, NULL, NULL, NULL)," +
"('linestatus', NULL, NULL, 0.0, NULL, NULL, NULL)," +
"('shipdate', NULL, NULL, 0.0, NULL, '1992-01-04', '1998-11-29')," +
"('commitdate', NULL, NULL, 0.0, NULL, '1992-02-02', '1998-10-28')," +
"('receiptdate', NULL, NULL, 0.0, NULL, '1992-01-09', '1998-12-25')," +
"('shipinstruct', NULL, NULL, 0.0, NULL, NULL, NULL)," +
"('shipmode', NULL, NULL, 0.0, NULL, NULL, NULL)," +
"('comment', NULL, NULL, 0.0, NULL, NULL, NULL)," +
"(NULL, NULL, NULL, NULL, 60175.0, NULL, NULL)");
"('nationkey', null, 25.0, 0.0, null, 0, 24)," +
"('regionkey', null, 5.0, 0.0, null, 0, 4)," +
"('comment', 1857.0, 25.0, 0.0, null, null, null)," +
"('name', 177.0, 25.0, 0.0, null, null, null)," +
"(null, null, null, null, 25.0, null, null)");
}

@Test
Expand Down Expand Up @@ -1364,7 +1360,10 @@ private void testDeltaLakeTableLocationChanged(boolean fewerEntries, boolean fir
public void testAnalyze()
{
String tableName = "test_analyze_" + randomNameSuffix();
assertUpdate("CREATE TABLE " + tableName
Session sessionWithDisabledStatisticsOnWrite = Session.builder(getSession())
.setCatalogSessionProperty(getSession().getCatalog().orElseThrow(), EXTENDED_STATISTICS_COLLECT_ON_WRITE, "false")
.build();
assertUpdate(sessionWithDisabledStatisticsOnWrite, "CREATE TABLE " + tableName
+ " WITH ("
+ "location = '" + getLocationForTable(bucketName, tableName) + "'"
+ ")"
Expand Down
100 changes: 80 additions & 20 deletions plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeAnalyze.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
import static io.trino.plugin.deltalake.DeltaLakeQueryRunner.DELTA_CATALOG;
import static io.trino.plugin.deltalake.DeltaLakeQueryRunner.TPCH_SCHEMA;
import static io.trino.plugin.deltalake.DeltaLakeQueryRunner.createDeltaLakeQueryRunner;
import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.EXTENDED_STATISTICS_COLLECT_ON_WRITE;
import static io.trino.testing.TestingAccessControlManager.TestingPrivilegeType.INSERT_TABLE;
import static io.trino.testing.TestingAccessControlManager.privilege;
import static io.trino.testing.TestingNames.randomNameSuffix;
Expand Down Expand Up @@ -71,17 +72,6 @@ private void testAnalyze(Optional<Integer> checkpointInterval)
+ (checkpointInterval.isPresent() ? format(" WITH (checkpoint_interval = %s)", checkpointInterval.get()) : "")
+ " AS SELECT * FROM tpch.sf1.nation", 25);

assertQuery(
"SHOW STATS FOR " + tableName,
"VALUES " +
"('nationkey', null, null, 0.0, null, 0, 24)," +
"('regionkey', null, null, 0.0, null, 0, 4)," +
"('comment', null, null, 0.0, null, null, null)," +
"('name', null, null, 0.0, null, null, null)," +
"(null, null, null, null, 25.0, null, null)");

runAnalyzeVerifySplitCount(tableName, 1);

assertQuery(
"SHOW STATS FOR " + tableName,
"VALUES " +
Expand Down Expand Up @@ -156,13 +146,13 @@ public void testAnalyzePartitioned()
assertQuery(
"SHOW STATS FOR " + tableName,
"VALUES " +
"('nationkey', null, null, 0.0, null, 0, 24)," +
"('nationkey', null, 25.0, 0.0, null, 0, 24)," +
"('regionkey', null, 5.0, 0.0, null, null, null)," +
"('comment', null, null, 0.0, null, null, null)," +
"('name', null, null, 0.0, null, null, null)," +
"('comment', 1857.0, 25.0, 0.0, null, null, null)," +
"('name', 177.0, 25.0, 0.0, null, null, null)," +
"(null, null, null, null, 25.0, null, null)");

runAnalyzeVerifySplitCount(tableName, 5);
runAnalyzeVerifySplitCount(tableName, 1);

assertQuery(
"SHOW STATS FOR " + tableName,
Expand Down Expand Up @@ -276,7 +266,10 @@ public void testAnalyzeWithFilesModifiedAfter()
{
String tableName = "test_analyze_" + randomNameSuffix();

assertUpdate("CREATE TABLE " + tableName + " AS SELECT * FROM tpch.sf1.nation", 25);
assertUpdate(
disableStatisticsCollectionOnWrite(getSession()),
"CREATE TABLE " + tableName + " AS SELECT * FROM tpch.sf1.nation",
25);

Thread.sleep(100);
Instant afterInitialDataIngestion = Instant.now();
Expand Down Expand Up @@ -407,10 +400,6 @@ public void testDropExtendedStats()
+ "('name', 177.0, 25.0, 0.0, null, null, null),"
+ "(null, null, null, null, 25.0, null, null)";

assertQuery(query, baseStats);

// Update stats to include distinct count
runAnalyzeVerifySplitCount(table.getName(), 1);
assertQuery(query, extendedStats);

// Dropping extended stats clears distinct count and leaves other stats alone
Expand Down Expand Up @@ -479,6 +468,70 @@ public void testStatsOnTpcDsData()
}
}

@Test
public void testCreateTableStatisticsWhenCollectionOnWriteDisabled()
{
String tableName = "test_statistics_" + randomNameSuffix();
assertUpdate(
disableStatisticsCollectionOnWrite(getSession()),
"CREATE TABLE " + tableName + " AS SELECT * FROM tpch.sf1.nation",
25);

assertQuery(
"SHOW STATS FOR " + tableName,
"VALUES " +
"('nationkey', null, null, 0.0, null, 0, 24)," +
"('regionkey', null, null, 0.0, null, 0, 4)," +
"('comment', null, null, 0.0, null, null, null)," +
"('name', null, null, 0.0, null, null, null)," +
"(null, null, null, null, 25.0, null, null)");

assertUpdate("ANALYZE " + tableName);

assertQuery(
"SHOW STATS FOR " + tableName,
"VALUES " +
"('nationkey', null, 25.0, 0.0, null, 0, 24)," +
"('regionkey', null, 5.0, 0.0, null, 0, 4)," +
"('comment', 1857.0, 25.0, 0.0, null, null, null)," +
"('name', 177.0, 25.0, 0.0, null, null, null)," +
"(null, null, null, null, 25.0, null, null)");
}

@Test
public void testCreatePartitionedTableStatisticsWhenCollectionOnWriteDisabled()
{
String tableName = "test_statistics_" + randomNameSuffix();
assertUpdate(
disableStatisticsCollectionOnWrite(getSession()),
"CREATE TABLE " + tableName
+ " WITH ("
+ " partitioned_by = ARRAY['regionkey']"
+ ")"
+ "AS SELECT * FROM tpch.sf1.nation",
25);

assertQuery(
"SHOW STATS FOR " + tableName,
"VALUES " +
"('nationkey', null, null, 0.0, null, 0, 24)," +
"('regionkey', null, 5.0, 0.0, null, null, null)," +
"('comment', null, null, 0.0, null, null, null)," +
"('name', null, null, 0.0, null, null, null)," +
"(null, null, null, null, 25.0, null, null)");

assertUpdate("ANALYZE " + tableName);

assertQuery(
"SHOW STATS FOR " + tableName,
"VALUES " +
"('nationkey', null, 25.0, 0.0, null, 0, 24)," +
"('regionkey', null, 5.0, 0.0, null, null, null)," +
"('comment', 1857.0, 25.0, 0.0, null, null, null)," +
"('name', 177.0, 25.0, 0.0, null, null, null)," +
"(null, null, null, null, 25.0, null, null)");
}

private void runAnalyzeVerifySplitCount(String tableName, long expectedSplitCount)
{
MaterializedResultWithQueryId analyzeResult = getDistributedQueryRunner().executeWithQueryId(getSession(), "ANALYZE " + tableName);
Expand All @@ -502,4 +555,11 @@ private OperatorStats getOperatorStats(QueryId queryId)
.filter(summary -> summary.getOperatorType().contains("Scan"))
.collect(onlyElement());
}

private static Session disableStatisticsCollectionOnWrite(Session session)
{
return Session.builder(session)
.setCatalogSessionProperty(session.getCatalog().orElseThrow(), EXTENDED_STATISTICS_COLLECT_ON_WRITE, "false")
.build();
}
}
3 changes: 3 additions & 0 deletions plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,7 @@ public void testDefaults()
.setDynamicFilteringWaitTimeout(new Duration(0, SECONDS))
.setTableStatisticsEnabled(true)
.setExtendedStatisticsEnabled(true)
.setCollectExtendedStatisticsOnWrite(true)
.setCompressionCodec(HiveCompressionCodec.SNAPPY)
.setDeleteSchemaLocationsFallback(false)
.setParquetTimeZone(TimeZone.getDefault().getID())
Expand Down Expand Up @@ -93,6 +94,7 @@ public void testExplicitPropertyMappings()
.put("delta.dynamic-filtering.wait-timeout", "30m")
.put("delta.table-statistics-enabled", "false")
.put("delta.extended-statistics.enabled", "false")
.put("delta.extended-statistics.collect-on-write", "false")
.put("delta.compression-codec", "GZIP")
.put("delta.per-transaction-metastore-cache-maximum-size", "500")
.put("delta.delete-schema-locations-fallback", "true")
Expand Down Expand Up @@ -124,6 +126,7 @@ public void testExplicitPropertyMappings()
.setDynamicFilteringWaitTimeout(new Duration(30, MINUTES))
.setTableStatisticsEnabled(false)
.setExtendedStatisticsEnabled(false)
.setCollectExtendedStatisticsOnWrite(false)
.setCompressionCodec(HiveCompressionCodec.GZIP)
.setDeleteSchemaLocationsFallback(true)
.setParquetTimeZone(nonDefaultTimeZone().getID())
Expand Down
12 changes: 6 additions & 6 deletions plugin/trino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeDelete.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,16 +216,16 @@ public void testStatsAfterDelete()
4);
assertQuery("SHOW STATS FOR " + tableName,
"VALUES " +
"('a', null, null, 0.5, null, 1, 7)," +
"('b', null, null, 0.5, null, 3, 9)," +
"('c', null, null, 0.75, null, 5, 5)," +
"('a', null, 2.0, 0.5, null, 1, 7)," +
"('b', null, 2.0, 0.5, null, 3, 9)," +
"('c', null, 1.0, 0.75, null, 5, 5)," +
"(null, null, null, null, 4.0, null, null)");
assertUpdate("DELETE FROM " + tableName + " WHERE c IS NULL", 3);
assertQuery("SHOW STATS FOR " + tableName,
"VALUES " +
"('a', null, null, 0.0, null, 1, 1)," +
"('b', null, null, 0.0, null, 3, 3)," +
"('c', null, null, 0.0, null, 5, 5)," +
"('a', null, 1.0, 0.0, null, 1, 1)," +
"('b', null, 1.0, 0.0, null, 3, 3)," +
"('c', null, 1.0, 0.0, null, 5, 5)," +
"(null, null, null, null, 1.0, null, null)");
}

Expand Down
10 changes: 5 additions & 5 deletions ...rino-delta-lake/src/test/java/io/trino/plugin/deltalake/TestDeltaLakeTableStatistics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ public void testShowStatsForTableWithNullsInPartitioningColumn()
"VALUES " +
// column_name | data_size | distinct_values_count | nulls_fraction | row_count | low_value | high_value
"('pk', null, 1.0, 0.5, null, null, null)," +
"('val_col', null, null, 0.0, null, 23, 24)," +
"('val_col', null, 2.0, 0.0, null, 23, 24)," +
"(null, null, null, null, 2.0, null, null)");
}

Expand All @@ -96,7 +96,7 @@ public void testShowStatsForTableWithTwoPartitioningColumns()
// column_name | data_size | distinct_values_count | nulls_fraction | row_count | low_value | high_value
"('pk1', null, 1.0, 0.25, null, null, null)," +
"('pk2', null, 3.0, 0.0, null, null, null)," +
"('val_col', null, null, 0.0, null, 23, 26)," +
"('val_col', null, 4.0, 0.0, null, 23, 26)," +
"(null, null, null, null, 4.0, null, null)");
}

Expand All @@ -114,7 +114,7 @@ public void testShowStatsForPartitioningColumnThatOnlyHasNulls()
"VALUES " +
// column_name | data_size | distinct_values_count | nulls_fraction | row_count | low_value | high_value
"('pk1', 0.0, 0.0, 1.0, null, null, null)," +
"('val_col', null, null, 0.0, null, 23, 24)," +
"('val_col', null, 2.0, 0.0, null, 23, 24)," +
"(null, null, null, null, 2.0, null, null)");
}

Expand All @@ -136,7 +136,7 @@ public void testShowStatsForQueryWithWhereClause()
// column_name | data_size | distinct_values_count | nulls_fraction | row_count | low_value | high_value
"('pk1', null, 1.0, 0.0, null, null, null)," +
"('pk2', null, 3.0, 0.0, null, null, null)," +
"('val_col', null, null, 0.0, null, 23, 26)," +
"('val_col', null, 3.0, 0.0, null, 23, 26)," +
"(null, null, null, null, 3.0, null, null)");
}

Expand All @@ -148,7 +148,7 @@ public void testShowStatsForAllNullColumn()
"SHOW STATS FOR show_stats_with_null",
"VALUES " +
// column_name | data_size | distinct_values_count | nulls_fraction | row_count | low_value | high_value
"('col', 0.0, null, 1.0, null, null, null)," +
"('col', 0.0, 0.0, 1.0, null, null, null)," +
"(null, null, null, null, 1.0, null, null)");
}
}
Loading