mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-24 13:52:34 +08:00
feat: Add Concatenate and Merge operations to DataFrame Operations (#11601)
* add concatenate and merge operations on tables * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes (attempt 2/3) * add code rabbit suggestions * [autofix.ci] apply automated fixes * [autofix.ci] apply automated fixes --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Cristhian Zanforlin Lousa
committed by
GitHub
GitHub
parent
15b98537d7
commit
0cedf8a5a0
@ -222,14 +222,12 @@ class TestEdgeCases:
|
||||
assert list(result.columns) == list(sample_dataframe.columns)
|
||||
|
||||
def test_invalid_operation_format(self, component, sample_dataframe):
|
||||
"""Test with invalid operation format."""
|
||||
"""Test with invalid operation format raises error."""
|
||||
component.df = sample_dataframe
|
||||
component.operation = "Invalid String" # Not list format
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
# Should return original DataFrame
|
||||
assert len(result) == len(sample_dataframe)
|
||||
with pytest.raises(ValueError, match="Unsupported operation"):
|
||||
component.perform_operation()
|
||||
|
||||
def test_empty_dataframe(self, component):
|
||||
"""Test operations on empty DataFrame."""
|
||||
@ -280,6 +278,8 @@ class TestDynamicUI:
|
||||
"num_rows": {"show": False},
|
||||
"replace_value": {"show": False},
|
||||
"replacement_value": {"show": False},
|
||||
"merge_on_column": {"show": False},
|
||||
"merge_how": {"show": False},
|
||||
}
|
||||
|
||||
# Select Filter operation
|
||||
@ -305,6 +305,8 @@ class TestDynamicUI:
|
||||
"num_rows": {"show": False},
|
||||
"replace_value": {"show": False},
|
||||
"replacement_value": {"show": False},
|
||||
"merge_on_column": {"show": False},
|
||||
"merge_how": {"show": False},
|
||||
}
|
||||
|
||||
# Select Sort operation
|
||||
@ -330,6 +332,8 @@ class TestDynamicUI:
|
||||
"num_rows": {"show": True},
|
||||
"replace_value": {"show": True},
|
||||
"replacement_value": {"show": True},
|
||||
"merge_on_column": {"show": True},
|
||||
"merge_how": {"show": True},
|
||||
}
|
||||
|
||||
# Deselect operation (empty list)
|
||||
@ -350,6 +354,8 @@ class TestDynamicUI:
|
||||
assert updated_config["num_rows"]["show"] is False
|
||||
assert updated_config["replace_value"]["show"] is False
|
||||
assert updated_config["replacement_value"]["show"] is False
|
||||
assert updated_config["merge_on_column"]["show"] is False
|
||||
assert updated_config["merge_how"]["show"] is False
|
||||
|
||||
|
||||
class TestDataTypes:
|
||||
@ -385,6 +391,241 @@ class TestDataTypes:
|
||||
assert len(result) == 2 # "text" and "more_text"
|
||||
|
||||
|
||||
class TestConcatenateOperation:
|
||||
"""Test concatenate operation for combining multiple DataFrames."""
|
||||
|
||||
def test_concatenate_two_dataframes(self, component):
|
||||
"""Test concatenating two DataFrames vertically."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [3, 4], "name": ["Charlie", "Diana"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Concatenate", "icon": "combine"}]
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 4
|
||||
assert list(result["id"]) == [1, 2, 3, 4]
|
||||
assert list(result["name"]) == ["Alice", "Bob", "Charlie", "Diana"]
|
||||
|
||||
def test_concatenate_single_dataframe(self, component):
|
||||
"""Test concatenate with only one DataFrame returns it unchanged."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]}))
|
||||
|
||||
component.df = [df1]
|
||||
component.operation = [{"name": "Concatenate", "icon": "combine"}]
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 2
|
||||
assert list(result["id"]) == [1, 2]
|
||||
|
||||
def test_concatenate_different_row_counts(self, component):
|
||||
"""Test concatenating DataFrames with different row counts."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2, 3], "value": ["a", "b", "c"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [4, 5], "value": ["d", "e"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Concatenate", "icon": "combine"}]
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 5
|
||||
|
||||
|
||||
class TestMergeOperation:
|
||||
"""Test merge operation for joining DataFrames."""
|
||||
|
||||
def test_merge_inner_join(self, component):
|
||||
"""Test inner merge on common column."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2, 3], "name": ["Alice", "Bob", "Charlie"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [2, 3, 4], "city": ["NYC", "LA", "Chicago"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Merge", "icon": "merge"}]
|
||||
component.merge_on_column = "id"
|
||||
component.merge_how = "inner"
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 2 # Only ids 2 and 3 exist in both
|
||||
assert "name" in result.columns
|
||||
assert "city" in result.columns
|
||||
|
||||
def test_merge_outer_join(self, component):
|
||||
"""Test outer merge includes all records."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [2, 3], "city": ["NYC", "LA"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Merge", "icon": "merge"}]
|
||||
component.merge_on_column = "id"
|
||||
component.merge_how = "outer"
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 3 # ids 1, 2, 3
|
||||
|
||||
def test_merge_left_join(self, component):
|
||||
"""Test left merge keeps all records from first DataFrame."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2, 3], "name": ["Alice", "Bob", "Charlie"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [2, 4], "city": ["NYC", "Chicago"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Merge", "icon": "merge"}]
|
||||
component.merge_on_column = "id"
|
||||
component.merge_how = "left"
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 3 # All from df1
|
||||
|
||||
def test_merge_right_join(self, component):
|
||||
"""Test right merge keeps all records from second DataFrame."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [2, 3, 4], "city": ["NYC", "LA", "Chicago"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Merge", "icon": "merge"}]
|
||||
component.merge_on_column = "id"
|
||||
component.merge_how = "right"
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 3 # All from df2
|
||||
|
||||
def test_merge_single_dataframe_returns_original(self, component):
|
||||
"""Test merge with single DataFrame returns it unchanged."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]}))
|
||||
|
||||
component.df = [df1]
|
||||
component.operation = [{"name": "Merge", "icon": "merge"}]
|
||||
component.merge_on_column = "id"
|
||||
component.merge_how = "inner"
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 2
|
||||
|
||||
def test_merge_invalid_column_raises_error(self, component):
|
||||
"""Test merge with non-existent column raises ValueError."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [2, 3], "city": ["NYC", "LA"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Merge", "icon": "merge"}]
|
||||
component.merge_on_column = "non_existent"
|
||||
component.merge_how = "inner"
|
||||
|
||||
with pytest.raises(ValueError, match="not found in first DataFrame"):
|
||||
component.perform_operation()
|
||||
|
||||
def test_merge_same_columns_coalesces_values(self, component):
|
||||
"""Test merge with same columns uses coalesce (df1 value or df2 value)."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2], "value": ["a", "b"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [2, 3], "value": ["x", "y"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Merge", "icon": "merge"}]
|
||||
component.merge_on_column = "id"
|
||||
component.merge_how = "outer"
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 3
|
||||
# Check no duplicate columns with _df2 suffix
|
||||
assert "value_df2" not in result.columns
|
||||
# Verify coalesced values
|
||||
assert result.loc[result["id"] == 1, "value"].iloc[0] == "a" # from df1
|
||||
assert result.loc[result["id"] == 2, "value"].iloc[0] == "b" # from df1 (coalesced)
|
||||
assert result.loc[result["id"] == 3, "value"].iloc[0] == "y" # from df2
|
||||
|
||||
def test_merge_more_than_two_dataframes_raises_error(self, component):
|
||||
"""Test merge with more than 2 DataFrames raises ValueError."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1], "name": ["A"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [2], "name": ["B"]}))
|
||||
df3 = DataFrame(pd.DataFrame({"id": [3], "name": ["C"]}))
|
||||
|
||||
component.df = [df1, df2, df3]
|
||||
component.operation = [{"name": "Merge", "icon": "merge"}]
|
||||
component.merge_on_column = "id"
|
||||
component.merge_how = "inner"
|
||||
|
||||
with pytest.raises(ValueError, match="Merge requires exactly"):
|
||||
component.perform_operation()
|
||||
|
||||
|
||||
class TestListInputHandling:
|
||||
"""Test that component handles list inputs correctly."""
|
||||
|
||||
def test_operations_use_first_dataframe_from_list(self, component):
|
||||
"""Test that non-merge operations use only the first DataFrame."""
|
||||
df1 = DataFrame(pd.DataFrame({"id": [1, 2], "name": ["Alice", "Bob"]}))
|
||||
df2 = DataFrame(pd.DataFrame({"id": [3, 4], "name": ["Charlie", "Diana"]}))
|
||||
|
||||
component.df = [df1, df2]
|
||||
component.operation = [{"name": "Head", "icon": "arrow-up"}]
|
||||
component.num_rows = 1
|
||||
|
||||
result = component.perform_operation()
|
||||
|
||||
assert len(result) == 1
|
||||
assert result.iloc[0]["name"] == "Alice" # From first DataFrame
|
||||
|
||||
|
||||
class TestMergeDynamicUI:
|
||||
"""Test dynamic UI for Merge and Concatenate operations."""
|
||||
|
||||
def test_merge_fields_show(self, component):
|
||||
"""Test that merge fields show when Merge is selected."""
|
||||
build_config = {
|
||||
"column_name": {"show": False},
|
||||
"filter_value": {"show": False},
|
||||
"filter_operator": {"show": False},
|
||||
"ascending": {"show": False},
|
||||
"new_column_name": {"show": False},
|
||||
"new_column_value": {"show": False},
|
||||
"columns_to_select": {"show": False},
|
||||
"num_rows": {"show": False},
|
||||
"replace_value": {"show": False},
|
||||
"replacement_value": {"show": False},
|
||||
"merge_on_column": {"show": False},
|
||||
"merge_how": {"show": False},
|
||||
}
|
||||
|
||||
updated_config = component.update_build_config(build_config, [{"name": "Merge", "icon": "merge"}], "operation")
|
||||
|
||||
assert updated_config["merge_on_column"]["show"] is True
|
||||
assert updated_config["merge_how"]["show"] is True
|
||||
assert updated_config["column_name"]["show"] is False
|
||||
|
||||
def test_concatenate_hides_all_extra_fields(self, component):
|
||||
"""Test that Concatenate operation hides all extra fields."""
|
||||
build_config = {
|
||||
"column_name": {"show": True},
|
||||
"filter_value": {"show": True},
|
||||
"filter_operator": {"show": True},
|
||||
"ascending": {"show": True},
|
||||
"new_column_name": {"show": True},
|
||||
"new_column_value": {"show": True},
|
||||
"columns_to_select": {"show": True},
|
||||
"num_rows": {"show": True},
|
||||
"replace_value": {"show": True},
|
||||
"replacement_value": {"show": True},
|
||||
"merge_on_column": {"show": True},
|
||||
"merge_how": {"show": True},
|
||||
}
|
||||
|
||||
updated_config = component.update_build_config(
|
||||
build_config, [{"name": "Concatenate", "icon": "combine"}], "operation"
|
||||
)
|
||||
|
||||
# Concatenate doesn't need any extra fields
|
||||
assert updated_config["column_name"]["show"] is False
|
||||
assert updated_config["merge_on_column"]["show"] is False
|
||||
assert updated_config["merge_how"]["show"] is False
|
||||
|
||||
|
||||
# Integration test to verify all operators work together
|
||||
def test_all_filter_operators_comprehensive():
|
||||
"""Comprehensive test of all filter operators on the same dataset."""
|
||||
|
||||
File diff suppressed because one or more lines are too long
@ -1381,7 +1381,7 @@
|
||||
},
|
||||
"DataFrameOperations": {
|
||||
"versions": {
|
||||
"0.3.0": "904f4eaebccd"
|
||||
"0.3.0": "e2b4323d4ed5"
|
||||
}
|
||||
},
|
||||
"DynamicCreateData": {
|
||||
|
||||
@ -16,9 +16,11 @@ class DataFrameOperationsComponent(Component):
|
||||
|
||||
OPERATION_CHOICES = [
|
||||
"Add Column",
|
||||
"Concatenate",
|
||||
"Drop Column",
|
||||
"Filter",
|
||||
"Head",
|
||||
"Merge",
|
||||
"Rename Column",
|
||||
"Replace Value",
|
||||
"Select Columns",
|
||||
@ -31,8 +33,9 @@ class DataFrameOperationsComponent(Component):
|
||||
DataFrameInput(
|
||||
name="df",
|
||||
display_name="DataFrame",
|
||||
info="The input DataFrame to operate on.",
|
||||
info="The input DataFrame to operate on. Connect multiple DataFrames for merge or concatenate operations.",
|
||||
required=True,
|
||||
is_list=True,
|
||||
),
|
||||
SortableListInput(
|
||||
name="operation",
|
||||
@ -41,9 +44,11 @@ class DataFrameOperationsComponent(Component):
|
||||
info="Select the DataFrame operation to perform.",
|
||||
options=[
|
||||
{"name": "Add Column", "icon": "plus"},
|
||||
{"name": "Concatenate", "icon": "combine"},
|
||||
{"name": "Drop Column", "icon": "minus"},
|
||||
{"name": "Filter", "icon": "filter"},
|
||||
{"name": "Head", "icon": "arrow-up"},
|
||||
{"name": "Merge", "icon": "merge"},
|
||||
{"name": "Rename Column", "icon": "pencil"},
|
||||
{"name": "Replace Value", "icon": "replace"},
|
||||
{"name": "Select Columns", "icon": "columns"},
|
||||
@ -138,6 +143,22 @@ class DataFrameOperationsComponent(Component):
|
||||
dynamic=True,
|
||||
show=False,
|
||||
),
|
||||
StrInput(
|
||||
name="merge_on_column",
|
||||
display_name="Merge On Column",
|
||||
info="The column name to merge DataFrames on. Must exist in both DataFrames.",
|
||||
dynamic=True,
|
||||
show=False,
|
||||
),
|
||||
DropdownInput(
|
||||
name="merge_how",
|
||||
display_name="Merge Type",
|
||||
options=["inner", "outer", "left", "right"],
|
||||
value="inner",
|
||||
info="Type of merge: inner (intersection), outer (union), left, or right.",
|
||||
dynamic=True,
|
||||
show=False,
|
||||
),
|
||||
]
|
||||
|
||||
outputs = [
|
||||
@ -161,6 +182,8 @@ class DataFrameOperationsComponent(Component):
|
||||
"num_rows",
|
||||
"replace_value",
|
||||
"replacement_value",
|
||||
"merge_on_column",
|
||||
"merge_how",
|
||||
]
|
||||
for field in dynamic_fields:
|
||||
build_config[field]["show"] = False
|
||||
@ -201,18 +224,27 @@ class DataFrameOperationsComponent(Component):
|
||||
build_config["replacement_value"]["show"] = True
|
||||
elif operation_name == "Drop Duplicates":
|
||||
build_config["column_name"]["show"] = True
|
||||
elif operation_name == "Merge":
|
||||
build_config["merge_on_column"]["show"] = True
|
||||
build_config["merge_how"]["show"] = True
|
||||
|
||||
return build_config
|
||||
|
||||
def perform_operation(self) -> DataFrame:
|
||||
df_copy = self.df.copy()
|
||||
def _get_primary_dataframe(self) -> DataFrame:
|
||||
"""Get the first DataFrame from input (handles both single and list inputs)."""
|
||||
if isinstance(self.df, list):
|
||||
return self.df[0].copy() if self.df else DataFrame()
|
||||
return self.df.copy()
|
||||
|
||||
# Handle SortableListInput format for operation
|
||||
def perform_operation(self) -> DataFrame:
|
||||
df_copy = self._get_primary_dataframe()
|
||||
|
||||
# Handle SortableListInput format for operation (also supports legacy string format)
|
||||
operation_input = getattr(self, "operation", [])
|
||||
if isinstance(operation_input, list) and len(operation_input) > 0:
|
||||
op = operation_input[0].get("name", "")
|
||||
if isinstance(operation_input, list):
|
||||
op = operation_input[0].get("name", "") if operation_input else ""
|
||||
else:
|
||||
op = ""
|
||||
op = operation_input or ""
|
||||
|
||||
# If no operation selected, return original DataFrame
|
||||
if not op:
|
||||
@ -238,6 +270,10 @@ class DataFrameOperationsComponent(Component):
|
||||
return self.replace_values(df_copy)
|
||||
if op == "Drop Duplicates":
|
||||
return self.drop_duplicates(df_copy)
|
||||
if op == "Concatenate":
|
||||
return self.concatenate_dataframes()
|
||||
if op == "Merge":
|
||||
return self.merge_dataframes()
|
||||
msg = f"Unsupported operation: {op}"
|
||||
logger.error(msg)
|
||||
raise ValueError(msg)
|
||||
@ -311,3 +347,65 @@ class DataFrameOperationsComponent(Component):
|
||||
|
||||
def drop_duplicates(self, df: DataFrame) -> DataFrame:
|
||||
return DataFrame(df.drop_duplicates(subset=self.column_name))
|
||||
|
||||
def concatenate_dataframes(self) -> DataFrame:
|
||||
"""Concatenate multiple DataFrames vertically (stack rows)."""
|
||||
if not isinstance(self.df, list) or len(self.df) == 0:
|
||||
return self.df.copy() if self.df is not None else DataFrame()
|
||||
|
||||
# If only one DataFrame, return it
|
||||
if len(self.df) == 1:
|
||||
return self.df[0].copy()
|
||||
|
||||
# Concatenate all DataFrames vertically
|
||||
concatenated = pd.concat(self.df, ignore_index=True)
|
||||
return DataFrame(concatenated)
|
||||
|
||||
def merge_dataframes(self) -> DataFrame:
|
||||
"""Merge two DataFrames based on a common column (join operation)."""
|
||||
if not isinstance(self.df, list) or len(self.df) == 0:
|
||||
return self.df.copy() if self.df is not None else DataFrame()
|
||||
|
||||
# If only one DataFrame, return it
|
||||
if len(self.df) == 1:
|
||||
return self.df[0].copy()
|
||||
|
||||
# Merge requires exactly two DataFrames
|
||||
max_merge_inputs = 2
|
||||
if len(self.df) > max_merge_inputs:
|
||||
msg = f"Merge requires exactly {max_merge_inputs} DataFrames, got {len(self.df)}"
|
||||
raise ValueError(msg)
|
||||
|
||||
df1 = self.df[0].copy()
|
||||
df2 = self.df[1].copy()
|
||||
|
||||
merge_on = getattr(self, "merge_on_column", None)
|
||||
merge_how = getattr(self, "merge_how", "inner")
|
||||
|
||||
# If merge column specified, validate it exists in both DataFrames
|
||||
if merge_on:
|
||||
if merge_on not in df1.columns:
|
||||
msg = f"Column '{merge_on}' not found in first DataFrame. Available: {list(df1.columns)}"
|
||||
raise ValueError(msg)
|
||||
if merge_on not in df2.columns:
|
||||
msg = f"Column '{merge_on}' not found in second DataFrame. Available: {list(df2.columns)}"
|
||||
raise ValueError(msg)
|
||||
|
||||
merged = df1.merge(df2, on=merge_on, how=merge_how, suffixes=("", "_df2"))
|
||||
else:
|
||||
merged = df1.merge(df2, left_index=True, right_index=True, how=merge_how, suffixes=("", "_df2"))
|
||||
|
||||
# Combine duplicate columns: use df1 value if exists, otherwise df2 value
|
||||
cols_to_drop = []
|
||||
for col in merged.columns:
|
||||
if col.endswith("_df2"):
|
||||
original_col = col[:-4] # Remove "_df2" suffix
|
||||
if original_col in merged.columns:
|
||||
# Coalesce: use original if not null, otherwise use _df2
|
||||
merged[original_col] = merged[original_col].combine_first(merged[col])
|
||||
cols_to_drop.append(col)
|
||||
|
||||
if cols_to_drop:
|
||||
merged = merged.drop(columns=cols_to_drop)
|
||||
|
||||
return DataFrame(merged)
|
||||
|
||||
Reference in New Issue
Block a user