MERGE logical plan vs execution plan schema mismatch

[ion-elgreco]

Environment

Delta-rs version: 0.15.1

Binding: python

---

Bug

What happened:
We do a merge on a predicate id==id and only do when_not_matched_insert_all. This works with local tests. So just this, nothing special since it just dispatches to TableMerger and then to merge execute in rust:

import polars as pl
pldf = pl.DataFrame()
  pldf.write_delta(
      dt,
      mode="merge",
      delta_merge_options=delta_merge_options,
  )
  .when_not_matched_insert_all()
  .execute()

For context the table is partitioned on year_month and code.

However, we get this weird schema mismatch even though it works in the tests:

LogicalPlan schema: DFSchema { fields: [
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "col1", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } },
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "col2", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }, 
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "date", data_type: Timestamp(Microsecond, None), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }, 
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "text1", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }, 
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "text2", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }, 
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "_commit_version", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }, 
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "m_rec_inserted_utc", data_type: Timestamp(Microsecond, None), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }, 
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "year_month", data_type: Dictionary(UInt16, Utf8), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }, 
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "code", data_type: Dictionary(UInt16, Utf8), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }, 
 DFField { qualifier: Some(Bare { table: "t" }), field: Field { name: "__delta_rs_path", data_type: Utf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} } }]

 ExecutionPlan schema: Schema { fields: [
 Field { name: "col1", data_type: LargeUtf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "col2", data_type: LargeUtf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "date", data_type: Timestamp(Microsecond, None), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "text1", data_type: LargeUtf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "text2", data_type: LargeUtf8, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "_commit_version", data_type: Int32, nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "m_rec_inserted_utc", data_type: Timestamp(Microsecond, None), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "code", data_type: Dictionary(UInt16, Utf8), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "year_month", data_type: Dictionary(UInt16, Utf8), nullable: true, dict_id: 0, dict_is_ordered: false, metadata: {} }, 
 Field { name: "__delta_rs_path", data_type: Dictionary(UInt16, Utf8), nullable: false, dict_id: 0, dict_is_ordered: false, metadata: {} }], metadata: {} }

Maybe the main culprit here is, the the delta_rs_path in the logical plan is Utf8 while in the execution plan is a dictionary. @Blajda

Otherwise the LargeUtf8 vs Utf8 is still an issue? But seems less likely since it worked before

[Blajda]

Does the error in this case happen consistently? Would be nice to have the exact stack trace / error.
Since this data is partitioned on two columns another possibility is the the distinct partition value scan might have an issue.

[ion-elgreco]

@Blajda yeah happens consistently, I'll try to see if I find some time to reproduce it with an MRE

I can try giving you the full stack trace but I have to rename a bunch of stuff since it's confidential, I'll get back to on that tomorrow:)

[sylvanayelda]

I am seeing the exact same behavior. My table is partitioned on two columns. When I change it to be partitioned by a single column, the error no longer occurs.

[ion-elgreco]

@sylvanayelda which version do you see the issue? Could you try it on 0.14.0 as well

[sylvanayelda]

@ion-elgreco I was using version 0.15.1. But with 0.14.0, I see the same issue (although the __delta_rs_path is no longer listed in either the LogicalPlan or ExecutionPlan schemas).

[Blajda]

@sylvanayelda Are you merging using the polars interface?
Please also provide the schema of the table you are using.

[sylvanayelda]

@Blajda I'm not using polars. Here is a sample of my code:

from deltalake import DeltaTable, write_deltalake
import pyarrow as pa

# schema is a pyarrow schema
source_table = pa.Table.from_pylist(records, schema=schema)
target_table = DeltaTable(table_path, storage_options=storage_opts)
(
          target_table.merge(
              source=source_table,
              source_alias=SOURCE_ALIAS,
              target_alias=TARGET_ALIAS,
              predicate=get_predicate(),
              large_dtypes=False,
          )
          .when_not_matched_insert(updates=get_inserts(schema.names))
          .execute()
        )

The table has the following schema:

Schema(
    [
        Field(partition_col1, PrimitiveType("string"), nullable=True),
        Field(col2, PrimitiveType("string"), nullable=True),
        Field(col3, PrimitiveType("string"), nullable=True),
        Field(col4, PrimitiveType("long"), nullable=True),
        Field(col5, PrimitiveType("long"), nullable=True),
        Field(col6, PrimitiveType("string"), nullable=True),
        Field(col7, PrimitiveType("string"), nullable=True),
        Field(col8, PrimitiveType("long"), nullable=True),
        Field(col9, PrimitiveType("long"), nullable=True),
        Field(col10, PrimitiveType("long"), nullable=True),
        Field(col11, PrimitiveType("long"), nullable=True),
        Field(
            col12,
            ArrayType(PrimitiveType("long"), contains_null=True),
            nullable=True,
        ),
        Field(
            col13,
            ArrayType(PrimitiveType("long"), contains_null=True),
            nullable=True,
        ),
        Field(partition_date, PrimitiveType("string"), nullable=True),
    ]
)

I should point out that we are also storing the data in ADLS2. Could that be causing any issues here?