[SPARK-50092][SQL] Fix PostgreSQL connector behaviour for multidimensional arrays

[PetarVasiljevic-DB]

What changes were proposed in this pull request?

There is a bug introduced in this PR #46006. This PR fixed the behaviour for PostgreSQL connector for multidimensional arrays since we have mapped all arrays to 1D arrays.

This PR has introduced a bug for one case. Following scenario is broken:

• User has a table t1 on Postgres and does CTAS command to create table t2 with same data.
  PR 46006 is resolving the dimensionality of column by reading the metadata from pg_attribute table and attndims column.
• This query returns correct dimensionality for table t1, but for table t2 that is created via CTAS it returns 0 always. This leads to all of the arrays being mapped to 0-D array which is the type itself (for example int). This is a bug on Postgres side.
• As a solution, we can query array_ndims function on given column that will return the dimension of the column. It works for CTAS-like-created tables too. We can get the result of this function on first row of table.

This change is doing additional query to PG table to find the dimension of array column instead of querying metadata table as before. It might be more expensive but we are sending LIMIT 1 in query.

Also, there is one caveat. In PG, there is no dimensionality of array as all the arrays are 1D arrays (https://www.postgresql.org/docs/current/arrays.html#ARRAYS-DECLARATION). Therefore, if there is table with 2D array column, it is totally fine from PG side to insert 1D or 3D array in this column. This makes the read path on Spark problematic since we will get the dimension of array, for example 1 if the first record is 1D array, and then we will try to read 3D array later on which will fail. Vice versa also, getting dimension 3 and reading 1D array later on. The change that I propose is fine with this scenario since it already doesn't work in Spark. What my change implies is that user can get different error message depending on the dimensionality of first record in table (namely, for one table they can get the error message that the expected type is ARRAY<ARRAY> and for the other that it is ARRAY<INT>.

Why are the changes needed?

Bug fix.

Does this PR introduce any user-facing change?

No. It just fixes one case that doesn't work currently.

How was this patch tested?

New tests.

Was this patch authored or co-authored using generative AI tooling?

[PetarVasiljevic-DB]

@yaooqinn as an author of #46006, I would like to get your review here :).

[MaxGekk]

@yaooqinn as an author of #46006, I would like to get your review here :).

also cc @dongjoon-hyun as you reviewed the original PR.

[yaooqinn]

Does this approach still work when the resultset given by limit 1 is empty?

[cloud-fan]

Does this approach still work when the resultset given by limit 1 is empty?

This is a good point. Shall we keep the information schema query as a fallback when the table is empty?

[PetarVasiljevic-DB]

Does this approach still work when the resultset given by limit 1 is empty?

This is actually fine as we won't set arrayDimension in metadata. Later on, when reading we are doing something like

val dim = if (metadata.contains("arrayDimension")) {
  metadata.getLong("arrayDimension").toInt
} else {
  1
}

so we are always going to fallback to 1. This is also what happened so far, if metadata from pg for some reason doesn't contain info about dimensionality we fallback to 1. I can write tests though to make sure this works.

[yaooqinn]

If we execute a CTAS on the Spark side against an empty Postgres table with multidimensional array columns, the new approach now defaults the dimension value to 1, then we have a malformed table in that Spark catalog, which might cause subsequent data pipelines to fail. While the information schema approach was initially impacted by an 'upstream' bug, it now appears that we have introduced a bug of our own.

[PetarVasiljevic-DB]

@yaooqinn makes sense, thanks! I have updated the PR with fallback to the querying of metadata table.

[PetarVasiljevic-DB]

What is your concern here?

[PetarVasiljevic-DB]

@yaooqinn are you questioning the reasoning behind metadata not being correct for CTAS created tables or I misunderstood your question?

[yaooqinn]

Hi @PetarVasiljevic-DB,

The experiment I conducted shows that executing array_ndim on the same column can result in different values across different rows.

[PetarVasiljevic-DB]

@yaooqinn yes, this is is purely dependant on the value of the row so it is possible we get different values. I have mentioned in PR description that we can have 2D array and we are allowed to insert 1D or 3D elements. Therefore we can expect returned dimension to be 1 or 3 depending on the first row fetched, sorry if it was not clear enough.

I would say this is fine since spark doesn't allow variable length arrays so if there are 2 rows with different dimensionality in foreign table, read will fail with or without this change.

What I can propose as a better way of fixing this is maybe querying the metadata first, and if it returns 0, we can fallback to querying array_ndim.

[yaooqinn]

The outcome of array_dims is data-dependent and volatile. I don't believe we can depend on it to create stable jobs. Instead of experiencing occasional failures, I would prefer to fail at the outset.

[PetarVasiljevic-DB]

Maybe we're not on the same page here but this is what currently happens in Spark:

• If array column on Postgres has such values that dimension of these values across all rows is same then job will be stable.
• If array column on Postgres has such values that dimension of these values across all rows are not same, the job won't be stable.

With the proposed fix, behaviour stays the same:

• If array_ndims is same for all rows, job will be stable
• If array_ndims returns multiple values across all rows, job won't be stable.

Currently, jobs that are reading from CTAS tables will always fail. Have I misunderstood the term "stability of job" here?

[milastdbx]

If postgres array has row with different array dimensionality (array_dims returning different values for different rows) - then from Spark perspective there is really no correct answer to what array_dims we should use given that spark cannot process this rows anyway. Given that there is no correct answer, all answers are wrong answers, so returning rand() (which limit 1 basically does) is same as reading it from metadata, as for Spark these are all incorrect.

However, there is one scenario that could be better - but I don't believe this PR makes it worse, it just doesn't solve it, and this PR might decrease "stability" of job (it sometimes fail, sometimes succeed).

Lets say customer issues a query:

select id, array_col from table1
where id = 3

Today if we read from metadata, lets say we get value 2, and this value for id = 3 just accidentaly happens to be correct and query predictably succeeds.

Now lets say we have query

select id, array_col from table1
where id = 4

Now lets imagine that array_col for id = 4 has dimensionality 4 - which will cause this query to predicatbly fail.

However if we switch to new model we will get in scenario that sometimes where id = 3 fails, sometimes succeeds.

So in short we shouldn't look at array_col dimensionality on table level but rather table + predicate level

[PetarVasiljevic-DB]

I see and it makes sense. Thanks @milastdbx. @yaooqinn may I suggest that we at least fallback to the value of 1 if we read 0 from metadata? It is expected to read the ArrayType so dimensionality of array should be at least 1. So something like this:

metadata.putLong("arrayDimension", Math.max(1, rs.getLong(1)))

It doesn't cover all the arrays still, but I am expecting that most of the users are using 1D arrays so this would cover most of the cases. Also, it is stable.

And I see it as pure improvement not from the current code of PG dialect, but from the previous one, where we were reading all PG arrays (non-CTAS and CTAS) as 1D. With this code, we would support arrays (non-CTAS) of higher dimensions, but for the CTAS tables we would support only 1D arrays.

[yaooqinn]

It makes sense to me w/ 1 as the default dimension value or maybe we can fail directly when encountering 0. Also cc @cloud-fan

[PetarVasiljevic-DB]

Can we merge? Test failures don't seem related to this RP.

[yaooqinn]

Can you rebase master and retry the GA?

[PetarVasiljevic-DB]

Tests are passing now.

[yaooqinn]

Merged to master, thank you @PetarVasiljevic-DB