abhishekrb19 commented on code in PR #19510:
URL: https://github.com/apache/druid/pull/19510#discussion_r4058071898
##########
docs/development/extensions-contrib/iceberg.md:
##########
@@ -194,6 +194,34 @@ Example:
When `residualFilterMode` is set to `fail` and a residual filter is detected,
the job will fail with an error message indicating which filter expression
produced the residual. This helps ensure data quality by preventing unintended
rows from being ingested.
+## Arrow vectorized reader
+
+By default the Iceberg input source resolves the snapshot to a list of data
file paths and reads them through the `warehouseSource`. Setting
`useArrowReader` to `true` reads the table scan directly with Iceberg's
vectorized Arrow reader instead, which avoids the per-file input format layer.
+
+| Property | Description | Default |
+|----------|-------------|---------|
+| `useArrowReader` | Read the table with Iceberg's vectorized Arrow reader. |
`false` |
+| `arrowBatchSize` | Rows per Arrow batch. Only applies when `useArrowReader`
is `true`. | `1024` |
Review Comment:
Do we anticipate adding more arrow-specific configurable properties? If so,
it'd be cleaner to have an `arrowConfig` container or so that'll encapsulate
and group all the arrow reader props together.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/iceberg/IcebergRestCatalogIngestionTest.java:
##########
@@ -151,30 +170,39 @@ public void testIngestFromIcebergRestCatalog()
+ "\"namespace\":\"%s\","
+ "\"icebergCatalog\":{\"type\":\"rest\",\"catalogUri\":\"%s\","
+
"\"catalogProperties\":{\"io-impl\":\"org.apache.iceberg.hadoop.HadoopFileIO\"}},"
- + "\"warehouseSource\":{\"type\":\"local\"}}',\n"
+ + "%s%s}',\n"
+ " '{\"type\":\"parquet\"}',\n"
+ " '[{\"type\":\"string\",\"name\":\"event_time\"},"
+ "{\"type\":\"string\",\"name\":\"name\"},"
+ "{\"type\":\"long\",\"name\":\"value\"}]'\n"
+ " )\n"
+ ")\n"
+ "PARTITIONED BY ALL TIME",
- dataSource,
+ targetDataSource,
ICEBERG_TABLE_NAME,
ICEBERG_NAMESPACE,
- catalogUri
+ catalogUri,
+ warehouseSource,
+ arrowReaderOptions
);
final SqlTaskStatus taskStatus = msqApis.submitTaskSql(sql);
cluster.callApi().waitForTaskToSucceed(taskStatus.getTaskId(), overlord);
- cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
+ cluster.callApi().waitForAllSegmentsToBeAvailable(targetDataSource,
coordinator, broker);
cluster.callApi().verifySqlQuery(
"SELECT __time, \"name\", \"value\" FROM %s ORDER BY __time",
- dataSource,
+ targetDataSource,
"2024-01-01T00:00:00.000Z,alice,100\n"
+ "2024-01-01T01:00:00.000Z,bob,200\n"
+ "2024-01-01T02:00:00.000Z,charlie,300"
);
+
+ log.info(
+ "Iceberg REST catalog ingestion timing: useArrowReader[%s],
dataSource[%s], elapsedMs[%d]",
+ useArrowReader,
+ targetDataSource,
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos)
Review Comment:
nit: use a `StopWatch`
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/iceberg/IcebergRestCatalogIngestionTest.java:
##########
@@ -151,30 +170,39 @@ public void testIngestFromIcebergRestCatalog()
+ "\"namespace\":\"%s\","
+ "\"icebergCatalog\":{\"type\":\"rest\",\"catalogUri\":\"%s\","
+
"\"catalogProperties\":{\"io-impl\":\"org.apache.iceberg.hadoop.HadoopFileIO\"}},"
- + "\"warehouseSource\":{\"type\":\"local\"}}',\n"
+ + "%s%s}',\n"
+ " '{\"type\":\"parquet\"}',\n"
+ " '[{\"type\":\"string\",\"name\":\"event_time\"},"
+ "{\"type\":\"string\",\"name\":\"name\"},"
+ "{\"type\":\"long\",\"name\":\"value\"}]'\n"
+ " )\n"
+ ")\n"
+ "PARTITIONED BY ALL TIME",
- dataSource,
+ targetDataSource,
ICEBERG_TABLE_NAME,
ICEBERG_NAMESPACE,
- catalogUri
+ catalogUri,
+ warehouseSource,
+ arrowReaderOptions
);
final SqlTaskStatus taskStatus = msqApis.submitTaskSql(sql);
cluster.callApi().waitForTaskToSucceed(taskStatus.getTaskId(), overlord);
- cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator,
broker);
+ cluster.callApi().waitForAllSegmentsToBeAvailable(targetDataSource,
coordinator, broker);
cluster.callApi().verifySqlQuery(
"SELECT __time, \"name\", \"value\" FROM %s ORDER BY __time",
- dataSource,
+ targetDataSource,
"2024-01-01T00:00:00.000Z,alice,100\n"
+ "2024-01-01T01:00:00.000Z,bob,200\n"
+ "2024-01-01T02:00:00.000Z,charlie,300"
);
+
+ log.info(
+ "Iceberg REST catalog ingestion timing: useArrowReader[%s],
dataSource[%s], elapsedMs[%d]",
Review Comment:
Does the main code (outside of tests) actually log something like this? It
seems generally useful to have a log and/or metric for this.
##########
embedded-tests/src/test/java/org/apache/druid/testing/embedded/iceberg/IcebergRestCatalogIngestionTest.java:
##########
@@ -136,7 +139,23 @@ public void tearDownIceberg()
@Test
public void testIngestFromIcebergRestCatalog()
{
+ ingestFromIcebergRestCatalog(dataSource, false);
+ }
+
+ @Test
+ public void testIngestFromIcebergRestCatalogWithArrowReader()
+ {
+ ingestFromIcebergRestCatalog(dataSource + "_arrow", true);
Review Comment:
Thanks for adding these tests!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
