feat(etl): implement ColumnRenameTransformer

Add transformer for renaming columns in the data stream during ETL.
Supports case-insensitive column name matching and multiple renames.
This commit is contained in:
Joseph Doherty
2026-01-03 09:13:02 -05:00
parent f1b7809a45
commit 81cb0df6bf
2 changed files with 120 additions and 0 deletions
@@ -0,0 +1,51 @@
using System.Data;
namespace JdeScoping.DataSync.Etl.Transformers;
/// <summary>
/// A data transformer that renames specified columns in the data stream.
/// Columns are matched by name (case-insensitive).
/// </summary>
public class ColumnRenameTransformer : DataTransformerBase
{
private readonly Dictionary<string, string> _renames;
private string[]? _outputNames;
private Dictionary<string, int>? _nameToOrdinal;
/// <inheritdoc />
public override string TransformerName => $"RenameColumns:{_renames.Count}";
/// <summary>
/// Creates a new ColumnRenameTransformer that renames the specified columns.
/// </summary>
/// <param name="renames">Tuples of (OldName, NewName) for columns to rename (case-insensitive matching).</param>
public ColumnRenameTransformer(params (string OldName, string NewName)[] renames)
{
ArgumentNullException.ThrowIfNull(renames);
_renames = renames.ToDictionary(r => r.OldName, r => r.NewName, StringComparer.OrdinalIgnoreCase);
}
/// <inheritdoc />
protected override void OnInitialize(IDataReader source)
{
_outputNames = new string[source.FieldCount];
_nameToOrdinal = new Dictionary<string, int>(StringComparer.OrdinalIgnoreCase);
for (int i = 0; i < source.FieldCount; i++)
{
var originalName = source.GetName(i);
var outputName = _renames.TryGetValue(originalName, out var newName) ? newName : originalName;
_outputNames[i] = outputName;
_nameToOrdinal[outputName] = i;
}
}
/// <inheritdoc />
public override string GetName(int ordinal, IDataReader source) => _outputNames![ordinal];
/// <inheritdoc />
public override int GetOrdinal(string name, IDataReader source)
{
if (_nameToOrdinal!.TryGetValue(name, out var ordinal)) return ordinal;
throw new IndexOutOfRangeException($"Column '{name}' not found.");
}
}
@@ -0,0 +1,69 @@
using System.Data;
using JdeScoping.DataSync.Etl.Transformers;
using NSubstitute;
namespace JdeScoping.DataSync.Tests.Etl.Transformers;
public class ColumnRenameTransformerTests
{
[Fact]
public void GetName_ReturnsRenamedColumn()
{
var source = CreateMockReader(new[] { "OldName", "Other" });
var transformer = new ColumnRenameTransformer(("OldName", "NewName"));
var reader = transformer.Transform(source);
Assert.Equal("NewName", reader.GetName(0));
Assert.Equal("Other", reader.GetName(1));
}
[Fact]
public void GetOrdinal_FindsByNewName()
{
var source = CreateMockReader(new[] { "OldName", "Other" });
var transformer = new ColumnRenameTransformer(("OldName", "NewName"));
var reader = transformer.Transform(source);
Assert.Equal(0, reader.GetOrdinal("NewName"));
}
[Fact]
public void GetOrdinal_OldName_ThrowsIndexOutOfRange()
{
var source = CreateMockReader(new[] { "OldName", "Other" });
var transformer = new ColumnRenameTransformer(("OldName", "NewName"));
var reader = transformer.Transform(source);
Assert.Throws<IndexOutOfRangeException>(() => reader.GetOrdinal("OldName"));
}
[Fact]
public void FieldCount_Unchanged()
{
var source = CreateMockReader(new[] { "OldName", "Other" });
var transformer = new ColumnRenameTransformer(("OldName", "NewName"));
var reader = transformer.Transform(source);
Assert.Equal(2, reader.FieldCount);
}
[Fact]
public void MultipleRenames_AllApplied()
{
var source = CreateMockReader(new[] { "A", "B", "C" });
var transformer = new ColumnRenameTransformer(("A", "Alpha"), ("C", "Charlie"));
var reader = transformer.Transform(source);
Assert.Equal("Alpha", reader.GetName(0));
Assert.Equal("B", reader.GetName(1));
Assert.Equal("Charlie", reader.GetName(2));
}
private static IDataReader CreateMockReader(string[] columns)
{
var reader = Substitute.For<IDataReader>();
reader.FieldCount.Returns(columns.Length);
for (int i = 0; i < columns.Length; i++)
{
var index = i;
reader.GetName(index).Returns(columns[index]);
reader.GetOrdinal(columns[index]).Returns(index);
}
return reader;
}
}