test: Add slt for swapped cross join with conflicting metadata

datafusion/sqllogictest/src/test_context.rs

@@ -182,6 +182,7 @@ impl TestContext {
             "metadata.slt" | "arrow_field.slt" => {
                 info!("Registering metadata table tables");
                 register_metadata_tables(test_ctx.session_ctx());
+                register_conflicting_metadata_tables(test_ctx.session_ctx())
             }
             "union_function.slt" => {
                 info!("Registering table with union column");
@@ -765,3 +766,26 @@ fn register_async_abs_udf(ctx: &SessionContext) {
     let udf = AsyncScalarUDF::new(Arc::new(async_abs));
     ctx.register_udf(udf.into_scalar_udf());
 }
+
+fn register_conflicting_metadata_tables(ctx: &SessionContext) {
+    let schema_left =
+        Schema::new(vec![Field::new("a", DataType::Int32, false)]).with_metadata(
+            HashMap::from([(String::from("metadata_key"), String::from("left"))]),
+        );
+    let data_left =
+        Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6, 7, 8, 9, 10])) as ArrayRef;
+
+    let batch_left =
+        RecordBatch::try_new(Arc::new(schema_left), vec![Arc::new(data_left)]).unwrap();
+    ctx.register_batch("larger_table", batch_left).unwrap();
+
+    let schema_right =
+        Schema::new(vec![Field::new("b", DataType::Int32, false)]).with_metadata(
+            HashMap::from([(String::from("metadata_key"), String::from("right"))]),
+        );
+    let data_right = Arc::new(Int32Array::from(vec![1])) as ArrayRef;
+
+    let batch_right =
+        RecordBatch::try_new(Arc::new(schema_right), vec![Arc::new(data_right)]).unwrap();
+    ctx.register_batch("smaller_table", batch_right).unwrap();
+}

datafusion/sqllogictest/test_files/metadata.slt

@@ -520,3 +520,11 @@ NULL the id field
 
 statement ok
 drop table table_with_metadata;
+
+# Test that swapped cross-join preserves metadata on conflicting values on metadata.
+# The larger_table has 10 values, smaller_tables 1 value and the fields of each table
+# have conflicting metadata, same key different values See test:context.rs register_conflicting_metadata_tables
+query II
+select * from larger_table cross join smaller_table order by larger_table.a limit 1;
+----
+1 1