feat: add support for add_column with backfill

[fangbo]

Related issue: #32

To add column with backfill data, some config should be set.

1. Read _rowaddr and _fragid from Lance dataset

      spark.conf().set("spark.sql.lance.with_metadata", "true");

2. Use "overwrite" mode to write and set "backfill_columns" to specify the new columns to add

      df2.write()
          .mode("overwrite")
          .option("backfill_columns", "new_col")
          .saveAsTable(catalogName + ".default." + tableName);

[github-actions]

ACTION NEEDED
Lance follows the Conventional Commits specification for release automation.

The PR title and description are used as the merge commit message. Please update your PR title and description to match the specification.

For details on the error please inspect the "PR Title Check" action.

[qidian99]

should both of two keys, _frag_id and _rowaddr, be sorted here?

[fangbo]

The physical plan is:

== Physical Plan ==
CommandResult <empty>
   +- AppendData org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy$$Lambda$1782/1752894940@6f7b8ae1, com.lancedb.lance.spark.write.AddColumnsWrite@6c8d8b60
      +- AdaptiveSparkPlan isFinalPlan=true
         +- == Final Plan ==
            *(2) Sort [_rowaddr#11L ASC NULLS FIRST], false, 0
            +- AQEShuffleRead coalesced
               +- ShuffleQueryStage 0
                  +- Exchange hashpartitioning(_fragid#12, 200), REPARTITION_BY_COL, [plan_id=84]
                     +- *(1) Project [_rowaddr#11L, _fragid#12, id#8, (id#8 * 100) AS new_col#39]
                        +- *(1) ColumnarToRow
                           +- BatchScan add_column_table[id#8, _rowaddr#11L, _fragid#12] class com.lancedb.lance.spark.read.LanceScan RuntimeFilters: []
         +- == Initial Plan ==
            Sort [_rowaddr#11L ASC NULLS FIRST], false, 0
            +- Exchange hashpartitioning(_fragid#12, 200), REPARTITION_BY_COL, [plan_id=68]
               +- Project [_rowaddr#11L, _fragid#12, id#8, (id#8 * 100) AS new_col#39]
                  +- BatchScan add_column_table[id#8, _rowaddr#11L, _fragid#12] class com.lancedb.lance.spark.read.LanceScan RuntimeFilters: []

The sort operator is after distribution. So I think we can only sort data by _rowaddr.

[qidian99]

Let's say the data is first repartitioned by _fragid, which guarantees that all rows from the same fragment end up in the same partition.
But if the sort doesn’t include _fragid, rows inside that partition may get interleaved? E.g., (1,0), (2,0), (1,1), (2,2)

[fangbo]

But if the sort doesn’t include _fragid, rows inside that partition may get interleaved? E.g., (1,0), (2,0), (1,1), (2,2)

_rowaddr is u64 and is composed by:

 frag_id (32) + row_index(32)

So I think rows will not get interleaved.

[jackye1995]

I am a bit hesitated about this user experience... we are clearly inventing something new here which is great, but overloading the overwrite mode feels wrong to me. So far the "norm" is that we can invent something in SQL, and then eventually dataframe operations will catch up. For example there was MERGE INTO SQL, and in Spark 4.0 now there is merge DataFrame operations. So to me having a SQL extensions feels "official" compared to using write overwrite that feels much more like a hack.

df2.createOrReplaceTempView("backfill_data")
spark.sql("ALTER TABLE ADD COLUMNS col1, col2 AS SELECT * FROM backfill_data")

I guess that brings the complexity of needing to add the SQL extensions and related parsers, but we probably need that anyway for features like compaction.

[jackye1995]

Another thing is that compared to this, there is technically also the add column using function experience, which would be harder to express in DataFrame, but we can invent SQL like:

spark.sql("ALTER TABLE ADD COLUMNS col1 AS my_udf(col2)")

[qidian99]

I agree, that custom DSL/SQL will deliver a better user experience. For now, changing the semantics of overwrite/mergeinto seems okay to me.

For the UDF scenario, if the computed column only depends on the current dataset (and is 1:1 row-aligned), the UDF approach is more intuitive. It's worth noting that there are also cases where we need to combine external data to compute new columns — e.g., join with external tables / lookup services -- where the DataFrame style offers greater expressiveness.

[fangbo]

I greatly appreciate your suggestion. I also think using SQL extensions feels more official.

[jackye1995]

shouldn't the row address naturally be ascending within a fragment? I don't think we need to force ordering it?

[fangbo]

shouldn't the row address naturally be ascending within a fragment? I don't think we need to force ordering it?

The data in the original table may be processed in a distributed manner, which can change the order of the records. So it is necessary to force order it by _rowaddr.

[fangbo]

@jackye1995 @qidian99 This PR is ready. Could you please review it? Thank you.