szehon-ho commented on code in PR #59683: URL: https://github.com/apache/spark/pull/59683#discussion_r4383912893
########## docs/sql-v2-data-sources.md: ########## @@ -466,6 +512,222 @@ The same `Table` and `Scan` abstractions support streaming queries. A `Table` th A `Table` that declares `STREAMING_WRITE` supports streaming writes through `Write.toStreaming()`, which returns a `StreamingWrite`. +## Building a Connector + +Spark's own test connectors make a working example of each step: +[`JavaSimpleDataSourceV2`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleDataSourceV2.java) +for a minimal read, +[`JavaAdvancedDataSourceV2WithV2Filter`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaAdvancedDataSourceV2WithV2Filter.java) +for pushdown, and +[`JavaSimpleWritableDataSource`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleWritableDataSource.java) +for writes, exercised by +[`DataSourceV2Suite`](https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala). +The three share a few test helpers. `TestingV2Source` is a test trait over `TableProvider`: it +implements `inferSchema` (two `int` columns, `i` and `j`) and forwards +`getTable(schema, partitioning, properties)` to a one-argument `getTable(options)`. The excerpts +override the one-argument form; a connector implements the three-argument form. `JavaSimpleBatchTable` is a +`Table` that declares `BATCH_READ` and returns that schema (through the deprecated `schema()`; a +new `Table` implements `columns()`); `JavaRangeInputPartition` carries a `start` and `end`; and +`JavaSimpleScanBuilder` is a `ScanBuilder` that is also its own `Scan` and `Batch`. + +### Step 1: A Minimal Batch Read + +The smallest connector is a `TableProvider` that returns a `Table`, a `Table` that declares +`BATCH_READ` and returns a `ScanBuilder`, and a `ScanBuilder` whose `Batch` plans partitions and +creates readers. `JavaSimpleDataSourceV2` is exactly that: + +{% highlight java %} +public class JavaSimpleDataSourceV2 implements TestingV2Source { + + static class MyScanBuilder extends JavaSimpleScanBuilder { + @Override + public InputPartition[] planInputPartitions() { + InputPartition[] partitions = new InputPartition[2]; + partitions[0] = new JavaRangeInputPartition(0, 5); + partitions[1] = new JavaRangeInputPartition(5, 10); + return partitions; + } + } + + @Override + public Table getTable(CaseInsensitiveStringMap options) { + return new JavaSimpleBatchTable() { + @Override + public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) { + return new MyScanBuilder(); + } + }; + } +} +{% endhighlight %} + +`planInputPartitions()` runs once on the driver and returns one `InputPartition` per task. Each +partition is serialized to an executor, so it should hold only what identifies the split (here a +range of integers; for a file source, a path and a byte range). The `PartitionReaderFactory` is +serialized the same way and creates a `PartitionReader` on the executor. The reader is a cursor: +`next()` advances, `get()` returns the current row, `close()` releases any handle. The one in +`JavaSimpleReaderFactory` counts through the partition's range: + +{% highlight java %} +return new PartitionReader<InternalRow>() { + private int current = p.start - 1; + + @Override + public boolean next() { + current += 1; + return current < p.end; + } + + @Override + public InternalRow get() { + return new GenericInternalRow(new Object[] {current, -current}); + } + + @Override + public void close() { + } +}; +{% endhighlight %} + +A `PartitionReader<InternalRow>` produces `InternalRow`, which holds each column +in Spark's internal representation rather than the external one a `Row` carries: a boxed `int` +is fine for an `IntegerType` column, but a `StringType` column holds `UTF8String` (not +`java.lang.String`), `DecimalType` holds `Decimal`, `DateType` holds an `int` of days, +`TimestampType` a `long` of microseconds, `BinaryType` a `byte[]`, and nested types hold +`ArrayData`, `MapData` and `InternalRow`. `InternalRow` and `GenericInternalRow` live in Spark's +internal `catalyst` package and have no API docs, but they are the row types the connector API +uses. The accessor names on `InternalRow` (`getUTF8String`, +`getDecimal`, `getInt`, `getLong`, `getBinary`, `getArray`, `getMap`, `getStruct`) are the +reference for each type. `GenericInternalRow` takes `Object[]`, so a value of the wrong +representation compiles and fails on the executor with a `ClassCastException`. + +With this much, `spark.read.format("...JavaSimpleDataSourceV2").load()` returns ten rows, and +the `select` and `filter` in the "simplest implementation" test work too: Spark applies the +projection and the filter itself, above the scan. + +### Step 2: Projection and Filter Pushdown (Optional) + +A connector that can skip columns or rows adds the pushdown mix-ins to its `ScanBuilder`. The +list, the order Spark applies them in, and the residual-filter contract are under +[ScanBuilder](#scanbuilder). `JavaAdvancedDataSourceV2WithV2Filter` accepts only +`i > <int literal>` and returns everything else for Spark to evaluate. Unlike the test, which +asserts the shapes it expects, the following version checks them, because Spark can also push `i > j` +as a `>` predicate; it also matches on the public `NamedReference` rather than the test's +`FieldReference`. It accepts a threshold only from `-1` up, because the planner starts the first +partition at the threshold plus one and the source's data is `0` to `9`; a lower threshold stays +with Spark. Replacing `predicates` on each call is right while `supportsIterativePushdown()` +returns `false` (the default); an iterative source accumulates them instead: + +{% highlight java %} +@Override +public Predicate[] pushPredicates(Predicate[] predicates) { + List<Predicate> supported = new ArrayList<>(); + List<Predicate> unsupported = new ArrayList<>(); + for (Predicate p : predicates) { + if (isIGreaterThanInt(p)) { + supported.add(p); Review Comment: [P2] Enforce every accepted predicate The snippet accepts every supported lower bound, but the linked AdvancedBatchWithV2Filter.planInputPartitions() uses the first and then breaks. If pushPredicates receives [i > 3, i > 6], both disappear from Spark's residual filters, while the reader returns 4..9 instead of 7..9. Compute the maximum accepted bound, or return predicates the planner does not enforce as residuals. ########## docs/sql-v2-data-sources.md: ########## @@ -466,6 +512,222 @@ The same `Table` and `Scan` abstractions support streaming queries. A `Table` th A `Table` that declares `STREAMING_WRITE` supports streaming writes through `Write.toStreaming()`, which returns a `StreamingWrite`. +## Building a Connector + +Spark's own test connectors make a working example of each step: +[`JavaSimpleDataSourceV2`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleDataSourceV2.java) +for a minimal read, +[`JavaAdvancedDataSourceV2WithV2Filter`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaAdvancedDataSourceV2WithV2Filter.java) +for pushdown, and +[`JavaSimpleWritableDataSource`](https://github.com/apache/spark/blob/master/sql/core/src/test/java/test/org/apache/spark/sql/connector/JavaSimpleWritableDataSource.java) +for writes, exercised by +[`DataSourceV2Suite`](https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala). +The three share a few test helpers. `TestingV2Source` is a test trait over `TableProvider`: it +implements `inferSchema` (two `int` columns, `i` and `j`) and forwards +`getTable(schema, partitioning, properties)` to a one-argument `getTable(options)`. The excerpts +override the one-argument form; a connector implements the three-argument form. `JavaSimpleBatchTable` is a +`Table` that declares `BATCH_READ` and returns that schema (through the deprecated `schema()`; a +new `Table` implements `columns()`); `JavaRangeInputPartition` carries a `start` and `end`; and +`JavaSimpleScanBuilder` is a `ScanBuilder` that is also its own `Scan` and `Batch`. + +### Step 1: A Minimal Batch Read + +The smallest connector is a `TableProvider` that returns a `Table`, a `Table` that declares +`BATCH_READ` and returns a `ScanBuilder`, and a `ScanBuilder` whose `Batch` plans partitions and +creates readers. `JavaSimpleDataSourceV2` is exactly that: + +{% highlight java %} +public class JavaSimpleDataSourceV2 implements TestingV2Source { + + static class MyScanBuilder extends JavaSimpleScanBuilder { + @Override + public InputPartition[] planInputPartitions() { + InputPartition[] partitions = new InputPartition[2]; + partitions[0] = new JavaRangeInputPartition(0, 5); + partitions[1] = new JavaRangeInputPartition(5, 10); + return partitions; + } + } + + @Override + public Table getTable(CaseInsensitiveStringMap options) { + return new JavaSimpleBatchTable() { + @Override + public ScanBuilder newScanBuilder(CaseInsensitiveStringMap options) { + return new MyScanBuilder(); + } + }; + } +} +{% endhighlight %} + +`planInputPartitions()` runs once on the driver and returns one `InputPartition` per task. Each +partition is serialized to an executor, so it should hold only what identifies the split (here a +range of integers; for a file source, a path and a byte range). The `PartitionReaderFactory` is +serialized the same way and creates a `PartitionReader` on the executor. The reader is a cursor: +`next()` advances, `get()` returns the current row, `close()` releases any handle. The one in +`JavaSimpleReaderFactory` counts through the partition's range: + +{% highlight java %} +return new PartitionReader<InternalRow>() { + private int current = p.start - 1; + + @Override + public boolean next() { + current += 1; + return current < p.end; + } + + @Override + public InternalRow get() { + return new GenericInternalRow(new Object[] {current, -current}); + } + + @Override + public void close() { + } +}; +{% endhighlight %} + +A `PartitionReader<InternalRow>` produces `InternalRow`, which holds each column +in Spark's internal representation rather than the external one a `Row` carries: a boxed `int` +is fine for an `IntegerType` column, but a `StringType` column holds `UTF8String` (not +`java.lang.String`), `DecimalType` holds `Decimal`, `DateType` holds an `int` of days, +`TimestampType` a `long` of microseconds, `BinaryType` a `byte[]`, and nested types hold +`ArrayData`, `MapData` and `InternalRow`. `InternalRow` and `GenericInternalRow` live in Spark's +internal `catalyst` package and have no API docs, but they are the row types the connector API +uses. The accessor names on `InternalRow` (`getUTF8String`, +`getDecimal`, `getInt`, `getLong`, `getBinary`, `getArray`, `getMap`, `getStruct`) are the +reference for each type. `GenericInternalRow` takes `Object[]`, so a value of the wrong +representation compiles and fails on the executor with a `ClassCastException`. + +With this much, `spark.read.format("...JavaSimpleDataSourceV2").load()` returns ten rows, and +the `select` and `filter` in the "simplest implementation" test work too: Spark applies the +projection and the filter itself, above the scan. + +### Step 2: Projection and Filter Pushdown (Optional) + +A connector that can skip columns or rows adds the pushdown mix-ins to its `ScanBuilder`. The +list, the order Spark applies them in, and the residual-filter contract are under +[ScanBuilder](#scanbuilder). `JavaAdvancedDataSourceV2WithV2Filter` accepts only +`i > <int literal>` and returns everything else for Spark to evaluate. Unlike the test, which +asserts the shapes it expects, the following version checks them, because Spark can also push `i > j` +as a `>` predicate; it also matches on the public `NamedReference` rather than the test's +`FieldReference`. It accepts a threshold only from `-1` up, because the planner starts the first +partition at the threshold plus one and the source's data is `0` to `9`; a lower threshold stays +with Spark. Replacing `predicates` on each call is right while `supportsIterativePushdown()` +returns `false` (the default); an iterative source accumulates them instead: + +{% highlight java %} +@Override +public Predicate[] pushPredicates(Predicate[] predicates) { + List<Predicate> supported = new ArrayList<>(); + List<Predicate> unsupported = new ArrayList<>(); + for (Predicate p : predicates) { + if (isIGreaterThanInt(p)) { + supported.add(p); + } else { + unsupported.add(p); + } + } + this.predicates = supported.toArray(new Predicate[0]); + return unsupported.toArray(new Predicate[0]); +} + +private static boolean isIGreaterThanInt(Predicate p) { + return p.name().equals(">") + && p.children()[0] instanceof NamedReference column + && column.fieldNames().length == 1 + && column.fieldNames()[0].equals("i") + && p.children()[1] instanceof Literal<?> value + && value.value() instanceof Integer bound + && bound >= -1; +} + +@Override +public Predicate[] pushedPredicates() { + return predicates; +} + +@Override +public void pruneColumns(StructType requiredSchema) { + this.requiredSchema = requiredSchema; +} +{% endhighlight %} + +`pushedPredicates()` reports what the source accepted. The `Batch` in the same class uses the +accepted predicate to plan fewer partitions, and its reader fills only the columns in +`requiredSchema`. The "advanced implementation with V2 Filter" test in +[`DataSourceV2Suite`](https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2Suite.scala) +checks the planned `Batch` for a few `filter` and `select` combinations. `EXPLAIN` does not show +`pushedPredicates()`; it shows `Scan.description()` inside the `BatchScan` node, after the output +columns. Its default is `getClass().toString()`, so override it to include the accepted predicates +and the pruned columns. + +### Step 3: Batch Writes + +A writable table declares `BATCH_WRITE` and implements `SupportsWrite`. For `SaveMode.Overwrite` +it also declares `TRUNCATE` (or `OVERWRITE_BY_FILTER`) and implements the matching builder mix-in, +such as `SupportsTruncate`; see +[`TableCapability`](api/java/org/apache/spark/sql/connector/catalog/TableCapability.html) for the +others. Spark checks the capability set during analysis, before calling any builder, so a missing +value fails the query even when the mix-in is implemented. `JavaSimpleWritableDataSource` declares +`BATCH_READ`, `BATCH_WRITE` and `TRUNCATE`. A `TableProvider` supports only the `append` and Review Comment: [P2] Qualify the save-mode restriction TableProvider implementations that also implement SupportsCatalogOptions support errorifexists and ignore: DataFrameWriter routes these through CTAS, and SupportsCatalogOptionsSuite covers both. Restrict this statement to the plain option-based provider shown here so it does not incorrectly rule out save modes supported by catalog-aware providers. -- 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]
