Full join on dataframe with only index yields dropped rows

[ntjohnson1]

Describe the bug
If I do a full join between a dataframe with content and one only consisting of an index column that index column only bits get dropped.

To Reproduce
See commented out empty additional column. When that is included then we see results in the final dataframe.

ctx = dfn.SessionContext()
 key_frame = ctx.from_pydict(
 {
 "log_time": [1, 3, 5, 7, 9, 11, 13, 15, 17, 19],
 "key_frame": [True, True, True, True, True, True, True, True, True, True]
 }
 )
 query_times = ctx.from_pydict(
 {
 "log_time": [2, 4, 6, 8, 10],
 #"empty": [0, 0, 0, 0, 0]
 }
 )

 print(key_frame)
 print(query_times)
 merged = query_times.join(key_frame, left_on="log_time", right_on="log_time", how="full")
 print(merged)

DataFrame()
+----------+-----------+
| log_time | key_frame |
+----------+-----------+
| 1 | true |
| 3 | true |
| 5 | true |
| 7 | true |
| 9 | true |
| 11 | true |
| 13 | true |
| 15 | true |
| 17 | true |
| 19 | true |
+----------+-----------+
DataFrame()
+----------+
| log_time |
+----------+
| 2 |
| 4 |
| 6 |
| 8 |
| 10 |
+----------+
DataFrame()
+----------+----------+-----------+
| log_time | log_time | key_frame |
+----------+----------+-----------+
| | 1 | true |
| | 3 | true |
| | 5 | true |
| | 7 | true |
| | 9 | true |
| | 11 | true |
| | 13 | true |
| | 15 | true |
| | 17 | true |
| | 19 | true |
+----------+----------+-----------+

Expected behavior
When doing a full join I get back all rows. Effectively merging the dataframes.

Here is a somewhat equivalent in pandas

key_frame_df = pd.DataFrame({
 "log_time": [1, 3, 5, 7, 9, 11, 13, 15, 17, 19],
 "key_frame": [True, True, True, True, True, True, True, True, True, True]
 })
 
 query_times_df = pd.DataFrame({
 "log_time": [2, 4, 6, 8, 10],
 # "empty": [0, 0, 0, 0, 0] # commented out like in original
 })
 
 
 # Perform full outer join (equivalent to DataFusion's "full" join)
 merged_df = pd.merge(query_times_df, key_frame_df, on="log_time", how="outer")
 print("\nMerged DataFrame (full outer join):")
 print(merged_df)

Merged DataFrame (full outer join):
 log_time key_frame
0 1 True
1 2 NaN
2 3 True
3 4 NaN
4 5 True
5 6 NaN
6 7 True
7 8 NaN
8 9 True
9 10 NaN
10 11 True
11 13 True
12 15 True
13 17 True
14 19 True

Actually the behavior in pyarrow is maybe a more direct comparison

key_table = pa.table(key_frame)
 query_table = pa.table(query_times)
 merged_table = query_table.join(key_table, keys="log_time", join_type="full outer")
 print(ctx.from_arrow(merged_table))

[IshaGudewar]

Thanks for the detailed report! I investigated the behavior and confirmed that the root cause is the schema merging during full outer joins.

When one of the input DataFrames contains only the join key (no extra columns), the join schema builder drops the duplicated join key column from that side. Since that side contributes no additional fields, the corresponding rows end up empty and are removed from the final output. This leads to incorrect results for full joins.

I can contribute by adding:

1. A regression test showing that a full join between
   df(log_time)
   df(log_time, key_frame)
   must return all rows from both sides.

2. A schema validation test verifying that join key rows from the “index-only” DataFrame are preserved.

If helpful, I can also explore patching the join schema builder to ensure a full join retains rows from both sides even when one side has only the join key.

I'll prepare a PR with the regression test shortly.

[renato2099]

I tried this in directly on datafusion, and it works as expected ... which is good 😅

 let a: ArrayRef = Arc::new(StringArray::from(vec!["a", "b", "c", "d"]));
 let b: ArrayRef = Arc::new(Int32Array::from(vec![1, 10, 10, 100]));
 let batch = RecordBatch::try_from_iter(vec![("a", a), ("b", b)])?;

 let bb: ArrayRef = Arc::new(Int32Array::from(vec![2, 20, 30, 40]));
 let batch2 = RecordBatch::try_from_iter(vec![("b", bb)])?;

 ctx.register_batch("t", batch)?;
 let df = ctx.table("t").await?;

 ctx.register_batch("t2", batch2)?;
 let df2 = ctx.table("t2").await?;

 df.join(df2, JoinType::Full, &["b"], &["b"], None)?.show().await?;

results

+---+-----+----+
| a | b | b |
+---+-----+----+
| a | 1 | |
| b | 10 | |
| c | 10 | |
| d | 100 | |
| | | 2 |
| | | 20 |
| | | 30 |
| | | 40 |
+---+-----+----+

also following your example, I see that all rows are there, but the values from the left side are dropped

>>> merged.to_pandas() 19:51:47 [50/903]
 log_time key_frame
0 NaN True
1 NaN True
2 NaN True
3 NaN True
4 NaN True
5 NaN True
6 NaN True
7 NaN True
8 NaN True
9 NaN True
10 2.0 None
11 4.0 None
12 6.0 None
13 8.0 None
14 10.0 None

and when using the empty column we get the following

>>> merged2.to_pandas()
 log_time empty key_frame
0 2.0 0.0 None
1 4.0 0.0 None
2 6.0 0.0 None
3 8.0 0.0 None
4 10.0 0.0 None
5 NaN NaN True
6 NaN NaN True
7 NaN NaN True
8 NaN NaN True
9 NaN NaN True
10 NaN NaN True
11 NaN NaN True
12 NaN NaN True
13 NaN NaN True
14 NaN NaN True

I am thinking that the left log_time column values are just getting dropped
I will take a closer look in the datafusion-python side :)

[renato2099]

thinking more about this , I think we should just not allow drop_duplicate_keys when doing a full outer join.
I think that would be the right behavior because
(1) For a FULL OUTER JOIN, the left and right join keys are NOT equivalent columns
(2) That would resemble the expected from upstream and what other databases do. For example, postgresql https://www.postgresql.org/docs/8.2/queries-table-expressions.html
(3) User can decide on how to merge columns afterwards.

then we can allow something like this:

batch = pa.RecordBatch.from_arrays(
 [pa.array([1, 2, 3]), pa.array(['a', 'b', 'c'])],
 names=["num", "name"],
 )
 batch2 = pa.RecordBatch.from_arrays(
 [pa.array([1, 3, 5]), pa.array([True, True, False])],
 names=["num", "value"],
 )

 df1 = ctx.create_dataframe([[batch]], "l")
 df2 = ctx.create_dataframe([[batch2]], "r")
 df3 = df1.join(df2, on="num", how="full", drop_duplicate_keys=False)
 df3.show()

+-----+------+-----+-------+
| num | name | num | value |
+-----+------+-----+-------+
| 1 | a | 1 | true |
| 3 | c | 3 | true |
| 2 | b | | |
| | | 5 | false |
+-----+------+-----+-------+

then user can do as they want, like coalescing

 df5 = df3.select_exprs(
 "coalesce(l.num, r.num) as num",
 "l.name",
 "r.value"
 )
 df5.show()

+-----+------+-------+
| num | name | value |
+-----+------+-------+
| 1 | a | true |
| 3 | c | true |
| 2 | b | |
| 5 | | false |
+-----+------+-------+

then no weird semantics + no workarounds + give user the power to decide what to do.

Also created PR for it here #1320

[renato2099]

hey @timsaucer @ntjohnson1 , would you mind sharing your thoughts on this?

[ntjohnson1]

I'm traveling all day today so I can't think carefully on this. I'd be fine with this approach if it didn't require the select_exprs since that feels a little unergonomic. The need to track the data frame names separately from the variable seems unfortunate.

I'd like to instead do df.select(F.coalesce(df1.col("num"), df2.col("num")).alias("num") unfortunately I don't think we currently have a syntax like that (or a nice alternative) to dedup columns that have the same name.

So if there was a way to resolve the columns without a specially crafted string I think the proposed direction seems reasonable.

[renato2099]

The need to track the data frame names separately from the variable seems unfortunate.

that is the case because of the way we are registering the dataframes with the SessionContext

 df1 = ctx.create_dataframe([[batch]], "l") # the dataframe is named "L" when referenced

In the example, we are doing this because we need it to disambiguate columns with the same name ... which is also our case 😅 (we have two columns that don't represent the same values but have the same name)

I'd like to instead do df.select(F.coalesce(df1.col("num"), df2.col("num")).alias("num") unfortunately I don't think we currently have a syntax like that (or a nice alternative) to dedup columns that have the same name.

right, unfortunately datafusion-python doesn't support that syntax atm (what is that btw? polars? spark-api?), but that is a good feature request though

we can do the following and no crafted string :)

 import datafusion.functions as F
 df6 = df3.select(
 F.coalesce(col("l.num"), col("r.num")).alias("num"),
 col("l.name"),
 col("r.value"),
 )
 df6.show()

then we get

+-----+------+-------+
| num | name | value |
+-----+------+-------+
| 1 | a | true |
| 3 | c | true |
| 2 | b | |
| 5 | | false |
+-----+------+-------+

[mesejo]

I've opened a PR that uses coalesce instead of dropping the left column. Using coalesce seems the right approach since:

1. It solves this issue (as suggested by the op @ntjohnson1)
2. Is in line with other major libraries such as pyarrow and polars

[renato2099]

I've opened a PR that uses coalesce instead of dropping the left column. Using coalesce seems the right approach since:

cool! thanks @mesejo ! yeah it is definitely an option :)

For me, it depends on how much SQL like we want/need to be + let the user decide the next operation (and hopefully an optimizer fuse operators together where it makes sense)

looking at how polars does it, it is interesting to see that columns get properly "renamed" such that users can also decide afterwards

import polars as pl

df = pl.DataFrame(
 {
 "foo": [1, 2, 3],
 "bar": [6.0, 7.0, 8.0],
 "ham": ["a", "b", "c"],
 }
)
other_df = pl.DataFrame(
 {
 "apple": ["x", "y", "z"],
 "ham": ["a", "b", "d"],
 }
)

out = df.join(other_df, on="ham", how="full")
print(out)

shape: (4, 5)
┌──────┬──────┬──────┬───────┬───────────┐
│ foo ┆ bar ┆ ham ┆ apple ┆ ham_right │
│ --- ┆ --- ┆ --- ┆ --- ┆ --- │
│ i64 ┆ f64 ┆ str ┆ str ┆ str │
╞══════╪══════╪══════╪═══════╪═══════════╡
│ 1 ┆ 6.0 ┆ a ┆ x ┆ a │
│ 2 ┆ 7.0 ┆ b ┆ y ┆ b │
│ null ┆ null ┆ null ┆ z ┆ d │
│ 3 ┆ 8.0 ┆ c ┆ null ┆ null │
└──────┴──────┴──────┴───────┴───────────┘

I haven't checked your PR yet @mesejo but if we are going that direction, we might as well consider column renaming (?)