LuciferYang commented on code in PR #14643:
URL: https://github.com/apache/gravitino/pull/14643#discussion_r4495446775
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java:
##########
@@ -31,14 +34,26 @@
import org.apache.spark.sql.connector.catalog.Identifier;
import org.apache.spark.sql.connector.write.LogicalWriteInfo;
import org.apache.spark.sql.connector.write.WriteBuilder;
+import org.apache.spark.sql.execution.datasources.jdbc.JDBCOptions;
+import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCScanBuilder;
import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTable;
import org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog;
+import org.apache.spark.sql.jdbc.JdbcDialects;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.util.CaseInsensitiveStringMap;
+import scala.collection.JavaConverters;
public class SparkJdbcTable extends JDBCTable {
+ private static final String[] UNSAFE_OPTIONS = {
+ "pushDownPredicate", "pushDownAggregate", "pushDownLimit", "pushDownOffset"
Review Comment:
This turns off predicate, aggregate, limit and offset pushdown for the whole
table as soon as it has one string, binary, UUID or external column, and on
Spark 3.5 as soon as a PostgreSQL table has a `timestamp` column. Collation,
binary literals and NTZ literals only matter for the expressions that contain
them.
I ran it through `GravitinoPostgreSqlCatalogSpark35/40` and
`GravitinoJdbcCatalogSpark35/40` on 3.5.3 and 4.0.3. On base, a 1M-row `(id
int, s text, n int)` table gets `WHERE id = 1`, `count(*) WHERE n > 500` and
`LIMIT 10` pushed; with this PR none are. PostgreSQL then receives `SELECT
"id","n" FROM "public"."big"` without the LIMIT, and at the default `fetchsize`
of 0 pgjdbc fetches the whole result before returning. A MySQL table with a
`varchar` column loses `id = 1` and `LIMIT 10` pushdown the same way. On 3.5, a
PostgreSQL table with only `int` and `timestamp` columns loses `WHERE ts >
TIMESTAMP_NTZ '2024-01-10 00:00:00'` and `max(ts)` pushdown, both of which base
pushes. Since the PR is labeled for branch-1.3, this would ship in a 1.3.x
patch release.
diqiu50 raised this earlier, and I think it belongs in this PR. I tried
returning a `JDBCScanBuilder` subclass from `newScanBuilder`. Its
`pushPredicates`, `pushAggregation` and `pushTopN` reject expressions that
involve a string or binary value (column, cast or literal, including array
elements) or an NTZ literal the dialect can't compile. LIMIT and OFFSET stay
on. On both versions `id = 1`, `LIMIT 10`, `count(*)` and `ORDER BY id LIMIT 3`
were pushed again, and string filters, string Top-N, `min/max(s)`, `cast(n AS
string) > '9'` and array ordering stayed in Spark. When a string predicate
stayed in Spark, LIMIT stayed in Spark with it. Aggregates need one more rule
for columns whose physical and Spark-visible types differ; see my comment on
line 161.
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java:
##########
@@ -44,7 +64,7 @@ public DataType toSparkType(Type gravitinoType) {
"Reading type {} as a string, Gravitino cannot represent it",
((Types.ExternalType) gravitinoType).catalogString());
return DataTypes.StringType;
- } else if (gravitinoType instanceof Types.TimestampType) {
+ } else if (gravitinoType instanceof Types.TimestampType &&
!useTimestampNtz) {
Review Comment:
`timestamp[]` columns can no longer be read on Spark 3.5. The parent
converter recurses into this method for `ListType`, so the element becomes NTZ
too. On 3.5.3, though, PostgresDialect maps an array element `timestamp` to
`TimestampType` and the array getter has no NTZ case, so the elements stay
`java.sql.Timestamp`. Base reads the column; with this PR 3.5.3 throws
`ClassCastException: class java.sql.Timestamp cannot be cast to class
java.lang.Long`, and 4.0.3 works.
Taking the element type from `jdbcTable.schema()` fixes it. I tried it: the
column reads on both versions, the 4.0 value is the same in UTC and
America/Los_Angeles sessions, and the 3.5 value shifts with the session time
zone, as on base. Keeping array elements as `TimestampType` everywhere also
reads, but then the 4.0 value shifts with the session time zone while top-level
columns don't. Please cover `timestamp[]` in `TestSparkJdbcReadCorrectness`.
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTypeConverter.java:
##########
@@ -44,7 +64,7 @@ public DataType toSparkType(Type gravitinoType) {
"Reading type {} as a string, Gravitino cannot represent it",
((Types.ExternalType) gravitinoType).catalogString());
return DataTypes.StringType;
- } else if (gravitinoType instanceof Types.TimestampType) {
+ } else if (gravitinoType instanceof Types.TimestampType &&
!useTimestampNtz) {
Review Comment:
The NTZ read path loses the clamp for PostgreSQL `infinity`. The base
`TimestampType` getter goes through
`PostgresDialect.convertJavaTimestampToTimestamp`, which maps
`infinity`/`-infinity` to 9999-12-31 and 0001-01-01. The NTZ getter calls
`convertJavaTimestampToTimestampNTZ`, which is just `t.toLocalDateTime` in
3.5.3, 4.0.3 and Spark master. Through the PostgreSQL catalog, a `timestamp`
column holding `'infinity'` reads as 9999-12-31 on base, and with this PR the
query fails with `ArithmeticException: long overflow` on 3.5.3 and 4.0.3.
NTZ is on by default for every PostgreSQL catalog and the PR is labeled for
branch-1.3, so this needs handling before merge. Registering an extra dialect
from Gravitino doesn't fix it. I registered a second PostgreSQL dialect
overriding both conversions: `JdbcDialects.get` returned an
`AggregatedDialect`, which forwards neither conversion, so both my overrides
and the built-in clamp were skipped and the read overflowed with the NTZ
mapping and with the old `TimestampType` mapping. On 4.0 the
`AggregatedDialect` also compiled a `byte[]` literal as `X'01'` instead of
`'\x01'::bytea`. A Java subclass of the 4.0 `PostgresDialect` does replace the
built-in one, but the registry is JVM-wide, so the subclass handles every
PostgreSQL JDBC URL in the application.
I'd make the NTZ mapping a catalog option that keeps the old mapping by
default, and document `infinity` as a known limitation when it's on. The root
cause is Spark's `convertJavaTimestampToTimestampNTZ`, which should get the
same clamp upstream.
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java:
##########
@@ -99,4 +145,51 @@ public StructType schema() {
public Map<String, String> properties() {
return gravitinoTableInfoHelper.properties();
}
+
+ private static boolean hasUnsafePushDown(JDBCTable table, Table source) {
+ for (Column column : source.columns()) {
+ Type type = column.dataType();
+ if (type instanceof Types.StringType
+ || type instanceof Types.VarCharType
+ || type instanceof Types.FixedCharType
+ || type instanceof Types.BinaryType
+ || type instanceof Types.ExternalType
+ || type instanceof Types.UUIDType) {
+ return true;
+ }
+ }
+ for (StructField field : table.schema().fields()) {
Review Comment:
This check reads the physical schema (`table.schema()`), but pushed literals
are typed by the schema Spark sees. On Spark 3.5 the two differ for
`timestamptz`: the 3.5 PostgresDialect doesn't tell it apart from `timestamp`,
so with `preferTimestampNTZ=true` it is physically NTZ while Spark sees
`TimestampType`. On 3.5.3, a table with only `int` and `timestamptz` columns
loses filter and aggregate pushdown, even `WHERE id = 1`, which base pushes.
`testUnsafeReadPushdownIsDisabledOnlyForAffectedTables` builds the physical
schema from the converter's output, so it never hits this case.
Checking the Spark-visible type instead isn't enough, because pushed
aggregates are typed by the reader. With that check and everything else pushed,
`max(tz)` failed planning with an `UnresolvedException` on 3.5.3, and on 3.5.6
(after SPARK-50870) it returned `2024-01-02 08:00:00` instead of `2024-01-02
00:00:00`. Not pushing aggregates over columns whose physical and Spark-visible
types differ fixed both, and `WHERE id = 1` was pushed again. Please add a 3.5
test for a `timestamptz` table.
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/SparkJdbcTable.java:
##########
@@ -99,4 +145,51 @@ public StructType schema() {
public Map<String, String> properties() {
return gravitinoTableInfoHelper.properties();
}
+
+ private static boolean hasUnsafePushDown(JDBCTable table, Table source) {
+ for (Column column : source.columns()) {
+ Type type = column.dataType();
+ if (type instanceof Types.StringType
Review Comment:
The classification only looks at top-level types, so array columns such as
`text[]` don't count and the table keeps all four pushdowns. PostgreSQL
compares arrays with the element collation, so the problem this PR fixes is
still there for array columns. On 3.5.3 and 4.0.3, with a `text[] COLLATE
"en-US-x-icu"` column holding `{a}` (id 1) and `{B}` (id 2), the pushed `ORDER
BY tags LIMIT 1` returns id 1 and the pushed `max(tags)` returns `{B}`;
evaluated in Spark they return id 2 and `{a}`. Base behaves the same, so this
is a gap in the new check rather than a regression.
Whether the check stays per table or moves per expression, it needs to
recurse into the `ListType` element type. Please add a `text[]` case to
`testUnsafeReadPushdownIsDisabledOnlyForAffectedTables`, or to the
per-expression test if the check moves.
##########
docs/spark-connector/spark-connector.md:
##########
@@ -152,3 +152,24 @@ these columns are rejected before JDBC execution,
including writes of NULL value
INSERT OVERWRITE. The error identifies the unsupported column. Spark binds
these columns
as VARCHAR, which does not preserve the PostgreSQL target type. Use a native
PostgreSQL
writer for these tables. Other PostgreSQL column types retain their existing
write behavior.
+
+### JDBC query consistency
+
+Some JDBC tables are read without pushdown. For these tables, Spark evaluates
predicates,
+aggregates, Top-N, limits, and offsets itself. This avoids result differences
caused by the
+column types below. A table is affected when it has a column of one of these
types:
+
+- String, char, varchar, binary, UUID, or an external type. The source
database can compare
+ strings with a different collation, and Spark cannot always write binary
literals in SQL.
+- A column that maps to Spark `TIMESTAMP_NTZ` (today, only a PostgreSQL
timestamp without time
+ zone), when the JDBC dialect cannot write `TIMESTAMP_NTZ` literals in SQL.
This is the case
+ in Spark 3.5, not in Spark 4.0.
+
+The rule applies to the whole table. Filters on numeric columns are not pushed
down either, so
+Spark can read more data from the source. You cannot turn pushdown back on
with the
+`pushDownPredicate`, `pushDownAggregate`, `pushDownLimit`, or `pushDownOffset`
options. Other
+tables keep their normal pushdown settings.
+
+For PostgreSQL catalogs, a timestamp without time zone maps to Spark
`TIMESTAMP_NTZ`. The
Review Comment:
Besides the read change this section describes, the NTZ mapping changes two
existing behaviors it doesn't mention. On 3.5.3 and 4.0.3, `partitionColumn` on
a PostgreSQL `timestamp` column now fails analysis with `Partition column type
should be numeric, date, or timestamp, but timestamp_ntz found`, while base
reads the table. Writes change too: with the JVM in Asia/Shanghai, inserting
`TIMESTAMP '2024-01-01 00:00:00'` from a UTC session stores 08:00 on base and
00:00 with this PR. From an Asia/Shanghai session both store 00:00.
Both should go into the docs and the release notes for the next 1.3.x
release. The write change only affects jobs whose session time zone differs
from the JVM time zone, but tables those jobs write to will mix rows stored in
the JVM time zone (before the upgrade) with rows stored in the session time
zone (after it). Users also have no working opt-out today: with
`preferTimestampNTZ=false` on the catalog the column is still `timestamp_ntz`
(see my comment on `GravitinoJdbcCatalog` line 57).
##########
spark-connector/spark-common/src/main/java/org/apache/gravitino/spark/connector/jdbc/GravitinoJdbcCatalog.java:
##########
@@ -53,6 +53,9 @@ protected TableCatalog createAndInitSparkCatalog(
Map<String, String> all =
getPropertiesConverter().toSparkCatalogProperties(options, properties);
applyJdbcCredential(gravitinoCatalogClient, all);
+ if (useTimestampNtz()) {
+ all.put("preferTimestampNTZ", "true");
Review Comment:
A user's `preferTimestampNTZ=false` can win over this `put`, while the
converter still maps `timestamp` to NTZ. User options arrive through
`CaseInsensitiveStringMap` with lowercased keys, so the reader gets both
`prefertimestampntz` and `preferTimestampNTZ`. I set `preferTimestampNTZ=false`
on the PostgreSQL catalog. On 3.5.3 (Scala 2.12) this `put` won; on 4.0.3 and
on 3.5.3 (Scala 2.13) the user's `false` won while the schema still showed
`timestamp_ntz`.
With the reader at `false`, the physical schema has no NTZ, so the NTZ check
in `SparkJdbcTable` doesn't fire and the table keeps pushdown on 4.0.3 and on
3.5.3 (Scala 2.13). Some pushed queries then return wrong results or fail. With
the JVM in Asia/Shanghai and the session in UTC, a pushed `max(ts)` on 4.0.3
returned `2024-01-19 16:00:00` instead of `2024-01-20 00:00:00`, the shift
#14642 is about. On 3.5.3 (Scala 2.13), `max(ts)` failed planning, and `WHERE
ts > TIMESTAMP_NTZ '2024-01-10 00:00:00'` was pushed unquoted and rejected by
PostgreSQL.
If you make the NTZ mapping a catalog option, as I suggested in my
`infinity` comment on `SparkJdbcTypeConverter` line 67, set the reader flag and
the converter from that one option. If it stays always on, remove any case
variant of the key before the `put`, as `safeOptions` does.
--
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]
