Question: connecting to old SQL server about etl.net HOT 15 OPEN

using System.Data;
using System.Data.Odbc;
using Paillave.Etl.Core;
using Paillave.Etl.FileSystem;
using Paillave.Etl.TextFile;
using PocEtlDotNet.Entities;

namespace PocEtlDotNet;
class Program
{
    static async Task Main(string[] args)
    {
        var connectionString1 = @"Driver={SQL Server};Server=MyServer;Database=Db1;Trusted_Connection=yes;";
        var connectionString2 = @"Driver={SQL Server};Server=MyServer;Database=Db2;Trusted_Connection=yes;";

        var processRunner = StreamProcessRunner.Create<string>(DefineProcess);
        var executionOptions = new ExecutionOptions<string>
        {
            Resolver = new SimpleDependencyResolver()
                            .Register<IDbConnection>(new OdbcConnection(connectionString1), "source1")
                            .Register<IDbConnection>(new OdbcConnection(connectionString2), "source2"),
            UseDetailedTraces = true,
            NoExceptionOnError= false,
        };

        var res = await processRunner.ExecuteAsync("Start output to files", executionOptions);

        Console.Write(res.Failed ? "Failed" : "Succeeded");
    }

    private static void DefineProcess(ISingleStream<string> contextStream)
    {
        contextStream
            .CrossApply<object, MyTableEntity>("strange stuff", (fileValue, dependencyResolver, cancellationToken, push) =>
            {
                using var connection = dependencyResolver.Resolve<IDbConnection>("source1");
                connection.Open();
                using var command = new OdbcCommand("select * from dbo.my_table", (OdbcConnection)connection);
                using var reader = command.ExecuteReader();
                while (reader.Read())
                {
                    var values = new MyTableEntity() { code = reader.GetString(0), name = reader.GetString(1) };
                    push(values);
                }
            })
            .Do("print", o => Console.WriteLine(o.name))
            .Select("create row to save", i => new { i.name, i.code })
            .ToTextFileValue("to file", @"C:\temp\out_source1.csv", FlatFileDefinition.Create(f => new { name = f.ToColumn("Name"), code = f.ToColumn("Code") }).IsColumnSeparated('|'))
            .WriteToFile("save to file", i => i.Name);

        contextStream
            .CrossApply<object, MyTableEntity>("strange stuff", (fileValue, dependencyResolver, cancellationToken, push) =>
            {
                using var connection = dependencyResolver.Resolve<IDbConnection>("source2");
                connection.Open();
                using var command = new OdbcCommand("select * from dbo.my_table", (OdbcConnection)connection);
                using var reader = command.ExecuteReader();
                while (reader.Read())
                {
                    var values = new MyTableEntity() { code = reader.GetString(0), name = reader.GetString(1) };
                    push(values);
                }
            })
            .Do("print", o => Console.WriteLine(o.name))
            .Select("create row to save", i => new { i.name, i.code })
            .ToTextFileValue("to file", @"C:\temp\out_source2.csv", FlatFileDefinition.Create(f => new { name = f.ToColumn("Name"), code = f.ToColumn("Code") }).IsColumnSeparated('|'))
            .WriteToFile("save to file", i => i.Name);
    }
}

using System.Data;
using System.Data.Odbc;
using Paillave.Etl.Core;
using Paillave.Etl.FileSystem;
using Paillave.Etl.SqlServer;
using Paillave.Etl.TextFile;
using PocEtlDotNet.Entities;

namespace PocEtlDotNet;
class Program
{
    static async Task Main(string[] args)
    {
        var connectionString1 = @"Driver={SQL Server};Server=MyServer;Database=Source1;Trusted_Connection=yes;";
        var connectionString2 = @"Driver={SQL Server};Server=MyServer;Database=Source2;Trusted_Connection=yes;";

        using var conn1 = new OdbcConnection(connectionString1);
        using var conn2 = new OdbcConnection(connectionString2);
        conn1.Open();
        conn2.Open();

        var processRunner = StreamProcessRunner.Create<string>(DefineProcess);
        var executionOptions = new ExecutionOptions<string>
        {
            Resolver = new SimpleDependencyResolver()
                            .Register<IDbConnection>(conn1, "source1")
                            .Register<IDbConnection>(conn2, "source2"),
            UseDetailedTraces = true,
            NoExceptionOnError= false,
        };

        var res = await processRunner.ExecuteAsync("Start output to files", executionOptions);

        Console.Write(res.Failed ? "Failed" : "Succeeded");
    }

    private static void DefineProcess(ISingleStream<string> contextStream)
    {
        contextStream
            .CrossApplySqlServerQuery("select", o => o
                .FromQuery("select * from dbo.myTable")
                .WithMapping(i => new
                {
                    code = i.ToColumn("code"),
                    name = i.ToColumn("name")
                })
                , "source1")
            .Do("print", o => Console.WriteLine(o.name))
            .Select("create row to save", i => new { i.name, i.code })
            .ToTextFileValue("to file", @"C:\temp\out_source1.csv", FlatFileDefinition.Create(f => new { name = f.ToColumn("Name"), code = f.ToColumn("Code") }).IsColumnSeparated('|'))
            .WriteToFile("save to file", i => i.Name);

        var afd = contextStream
            .CrossApplySqlServerQuery("select", o => o
                .FromQuery("select * from dbo.myTable")
                .WithMapping(i => new
                {
                    code = i.ToColumn("code"),
                    name = i.ToColumn("name")
                })
                , "sourece2")
            .Do("print", o => Console.WriteLine(o.name))
            .Select("create row to save", i => new { i.name, i.code })
            .ToTextFileValue("to file", @"C:\temp\out_source2.csv", FlatFileDefinition.Create(f => new { name = f.ToColumn("Name"), code = f.ToColumn("Code") }).IsColumnSeparated('|'))
            .WriteToFile("save to file", i => i.Name);
    }
}

        contextStream
            .Select("Create a value", _ => new
            {
                code = "CD",
                name = "CD name"
            })
            .SqlServerSave("save to db", o => o.WithConnection("source1").ToTable("dbo.myTable")) ;

        contextStream
            .CrossApplySqlServerQuery("get max", a => a
                .FromQuery("SELECT 1000 as ID")
                .WithMapping(a => new { trans_id = a.ToColumn<int>("ID") }), "source1")
            .Select("build criteria", i => new { TransId = 100 })
            .Do("printeris", o => Console.WriteLine(o))
            .CrossApplySqlServerQuery("select with last", s => s
                .FromQuery("SELECT TOP 10 [trans_id] FROM [dbo].[temp_trans] WHERE [trans_id] <= @TransId")
                .WithMapping(a => new { transId = a.ToColumn<int>("trans_id") })
            , "source2")
            .Do("print", o => Console.WriteLine(o));

            Debug.Assert(m_action != null);
            if (m_action is Func<Task, TResult> func)
            {
                m_result = func(antecedent);
                return;
            }

Paillave.Etl.Core.JobExecutionException
  HResult=0x80131500
  Message=Job execution failed
  Source=Paillave.Etl
  StackTrace:
   at Paillave.Etl.Core.StreamProcessRunner`1.<>c__DisplayClass14_0.<ExecuteAsync>b__3(Task t)
   at System.Threading.Tasks.ContinuationResultTaskFromTask`1.InnerInvoke() in /_/src/libraries/System.Private.CoreLib/src/System/Threading/Tasks/TaskContinuation.cs:line 88
   at System.Threading.ExecutionContext.RunFromThreadPoolDispatchLoop(Thread threadPoolThread, ExecutionContext executionContext, ContextCallback callback, Object state) in /_/src/libraries/System.Private.CoreLib/src/System/Threading/ExecutionContext.cs:line 268
--- End of stack trace from previous location ---
   at System.Threading.ExecutionContext.RunFromThreadPoolDispatchLoop(Thread threadPoolThread, ExecutionContext executionContext, ContextCallback callback, Object state) in /_/src/libraries/System.Private.CoreLib/src/System/Threading/ExecutionContext.cs:line 293
   at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot, Thread threadPoolThread) in /_/src/libraries/System.Private.CoreLib/src/System/Threading/Tasks/Task.cs:line 2349
--- End of stack trace from previous location ---
   at PocEtlDotNet.Program.<Main>d__0.MoveNext() in C:\GIT\OFS\Source\PocEtlDotNet\Program.cs:line 32
   at PocEtlDotNet.Program.<Main>(String[] args)

  This exception was originally thrown at this call stack:
    System.Data.Odbc.OdbcParameterCollection.ValidateType(object)
    System.Data.Odbc.OdbcParameterCollection.Add(object)
    Paillave.Etl.SqlServer.SqlCommandValueProvider<TIn, TOut>.PushValues(TIn, System.Action<TOut>, System.Threading.CancellationToken, Paillave.Etl.Core.IExecutionContext)
    Paillave.Etl.Core.CrossApplyStreamNode<TIn, TOut>.CreateOutputStream.AnonymousMethod__2(System.Action<TOut>, System.Threading.CancellationToken)
    Paillave.Etl.Reactive.Core.DeferredPushObservable<T>.InternStart()

Inner Exception 1:
InvalidCastException: The OdbcParameterCollection only accepts non-null OdbcParameter type objects, not SqlParameter objects.