Program.cs

using Amazon;
using Amazon.Glue;
using Amazon.Glue.Model;
using Amazon.S3;
using Amazon.S3.Transfer;
using System.Data.OleDb;
using ParquetSharp;
using System.Collections;
using System.Data;
using System.Data.Odbc; 
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;

namespace SqlToRedshift
{
    class Program
    {
        public static string RedshiftConnStr = "DRIVER=Amazon Redshift (x64);";
        public static string bucketName = "";
        static string keyName = "";
        static string filePath = "";
        static string tblName = "";
        static string crtDestTbl = "";

        public static string schemaNameSQL = "dbo";
        public static string schemaNameAWS = "dbo_dev";
        static string getTblsRowsCount = String.Format(@"SET NOCOUNT ON;
SELECT SCHEMA_NAME(TBL.schema_id) AS SchemaName, TBL.name AS TableName, CAST(SUM(PART.rows) AS INT) AS RowCounts
FROM sys.tables TBL
INNER JOIN sys.partitions PART ON TBL.object_id = PART.object_id
INNER JOIN sys.indexes IDX ON PART.object_id = IDX.object_id
AND PART.index_id = IDX.index_id
WHERE IDX.index_id < 2 AND TBL.name NOT IN('sysdiagrams','t_sup_prop_liens_') AND TBL.temporal_type != 1 AND TBL.name LIKE 't_%' AND SCHEMA_NAME(TBL.schema_id) = '{0}'
GROUP BY TBL.schema_id, TBL.name
ORDER BY TBL.schema_id, TBL.name DESC;", schemaNameSQL);
        public static string logFileName = Environment.ProcessPath + ".txt";
        public static string IAMRole = "";
        static string ConnectionName = "";
        public static OleDbDataAdapter adapter = new OleDbDataAdapter();
        static void Main(string[] args)
        {
            bool allTables = false;
            bool foundTable = false;
            int tablesCount = 1;
            int rowsNum = 0;
            string MSSQLConnStr = @"Provider=MSOLEDBSQL19.1;Server=xyz,abc;Database=some_db;UID={0};PWD={1};ApplicationIntent=ReadOnly;Encrypt=True;TrustServerCertificate=True;PacketSize=16383;ConnectRetryCount=3;APP=" + System.Diagnostics.Process.GetCurrentProcess().ProcessName;
            Console.WriteLine(DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + " Start");
            Console.WriteLine("Environment Version: {0}", Environment.Version.ToString());
            Console.WriteLine("Log file: {0}", logFileName);
            // delete log file if exist
            System.IO.File.Delete(logFileName);
            System.Data.DataTable dtTables = new DataTable();
            System.Data.DataTable dtstgTables = new DataTable();

            // Get IP address
            IPAddress ipaddr = Dns.GetHostAddresses(Dns.GetHostName()).Where(address => address.AddressFamily == AddressFamily.InterNetwork).First();
            string ip = ipaddr.ToString();

            string jsonString = Amazon.Util.EC2InstanceMetadata.GetData("/iam/info");
            // Create a JsonNode DOM from a JSON string.
            System.Text.Json.Nodes.JsonNode infoNode = System.Text.Json.Nodes.JsonNode.Parse(jsonString)!;
            System.Text.Json.Nodes.JsonNode iamrolenode = infoNode!["InstanceProfileArn"]!;
            IAMRole = iamrolenode.ToJsonString().Trim('"').Replace("instance-profile", "role");

            // Get Redshift connection name
            if (ip.StartsWith("10.138.86"))
            {
                ConnectionName = "a"; // dev
            }
            else if (ip.StartsWith("10.138.15"))
            {
                ConnectionName = "b"; //pre-prod
            }
            else
            {
                throw new Exception("Unknown IP " + ip);
            }

            // Get Redshift connection strinng  // Data Catalog
            var client = new AmazonGlueClient();
            var request = new GetConnectionRequest
            {
                Name = ConnectionName,
                HidePassword = false  // IMPORTANT
            };
            // Block on the async call
            var response = client.GetConnectionAsync(request).GetAwaiter().GetResult();
            var props = response.Connection.ConnectionProperties;
            string[] tmparr = response.Connection.ConnectionProperties["JDBC_CONNECTION_URL"].Split(":");
            string RedshiftServer = tmparr[2].Remove(0, 2);
            string RedshiftDB = ((tmparr[3]).Split("/"))[1];
            string RedshiftUID = props["USERNAME"];
            string RedShiftPWD = props["PASSWORD"];
            string RedShiftPort = ((tmparr[3]).Split("/"))[0];
            OdbcConnectionStringBuilder builder = new OdbcConnectionStringBuilder(RedshiftConnStr);
            builder.Add("UID", RedshiftUID);
            builder.Add("Database", RedshiftDB);
            builder.Add("Server", RedshiftServer);
            builder.Add("pwd", RedShiftPWD);
            RedshiftConnStr = builder.ConnectionString;

            ConnectionName = "c";
            request = new GetConnectionRequest
            {
                Name = ConnectionName,
                HidePassword = false  // IMPORTANT
            };
            response = client.GetConnectionAsync(request).GetAwaiter().GetResult();
            props = response.Connection.ConnectionProperties;

            MSSQLConnStr = String.Format(MSSQLConnStr, props["USERNAME"], props["PASSWORD"]);
            OleDbConnection conn = new OleDbConnection(MSSQLConnStr);
            conn.Open();
            Console.WriteLine("Default MS SQL Server database is:" + conn.Database);
            adapter.SelectCommand = new OleDbCommand(getTblsRowsCount, conn);
            dtTables.BeginLoadData();
            adapter.Fill(dtTables);
            dtTables.EndLoadData();
            conn.Close();

            using (OdbcConnection connDest = new OdbcConnection(RedshiftConnStr))
            {
                connDest.Open();
                System.Data.Odbc.OdbcDataAdapter rsh_adapter = new OdbcDataAdapter("SELECT SUBSTRING(t.table_name, 8) AS tbl FROM information_schema.tables t WHERE t.table_schema = 'pp_stg' AND table_name LIKE 'rs_tbl_t%' ORDER BY t.table_name", connDest);
                dtstgTables.BeginLoadData();
                rsh_adapter.Fill(dtstgTables);
                dtstgTables.EndLoadData();    
            }

            Console.WriteLine("Enter full table name like dbo.tblName or press Enter to get all pp_stg tables:");
            string? FullTableName = Console.ReadLine();

            if (FullTableName == "")
            { // User press Enter
                allTables = true;
                foundTable = true;
                tablesCount = dtstgTables.Rows.Count;
            }
            else
            {
                allTables = false;
                if (FullTableName.StartsWith("dbo."))
                {
                    tblName = FullTableName.Substring(4);
                }
                else
                {
                    tblName = FullTableName;
                }
            }

            // Use current directory + Exports folder
            string outputPath = Path.Combine(Directory.GetCurrentDirectory(), "Exports");
            File.AppendAllText(logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + " outputPath = " + outputPath + Environment.NewLine));

            var exporter = new SqlToParquetExporter(MSSQLConnStr);

            if (!allTables)
            {
                //Console.WriteLine("!allTables 2");
                Console.WriteLine(DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + " Looking for " + tblName + " in " + dtTables.Rows.Count.ToString() + " tables");
                foreach (DataRow row in dtTables.Rows)
                {
                    int comparison = String.Compare(row["TableName"].ToString(), tblName, comparisonType: StringComparison.OrdinalIgnoreCase);
                    // Console.WriteLine( comparison.ToString() + "  " + row["TableName"].ToString());
                    if (comparison == 0)
                    {
                        foundTable = true;
                        rowsNum = (int)row["RowCounts"];
                        break;
                    }
                }
            }
            if (!foundTable)
            {
                Console.WriteLine(DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + " Table " + tblName + " not found!");
                Environment.Exit(0);
            }
            SetBucket();
            if (allTables)
            {
                int tblNum = 1;
                foreach (DataRow row in dtstgTables.Rows)
                {
                    tblName = row["tbl"].ToString();
                    System.Data.DataRow[] rows = dtTables.Select("TableName = '" + tblName + "'");
                    rowsNum = 0;
                    if (rows.Length > 0)
                    {
                        rowsNum = (int)rows[0]["RowCounts"];
                    }
                    Console.WriteLine(String.Format("Table {0}.{1} has {2} rows", schemaNameSQL, tblName, rowsNum));
                    File.AppendAllText(logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + String.Format(" Table {0}.{1} has {2} rows", schemaNameSQL, tblName, rowsNum) + Environment.NewLine));
                    if (rowsNum > 0)
                    {
                        Console.WriteLine(DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $" Call ExportTables for {tblName} {tblNum} of {tablesCount}");
                        File.AppendAllText(logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $" Call ExportTables for {tblName} {tblNum} of {tablesCount}" + Environment.NewLine));
                        exporter.ExportTables(tblName, outputPath);
                    }
                    tblNum++;
                }
            }
            else
            {
                // do we have schema?
                if (FullTableName.IndexOf('.') == -1)
                { FullTableName = String.Concat("dbo.", FullTableName); }
                Console.WriteLine(String.Format("Table {0}.{1} has {2} rows", schemaNameSQL, tblName, rowsNum));
                if (rowsNum > 0)
                {
                    exporter.ExportTables(tblName, outputPath);
                    //rslt = ProcessTable(FullTableName, rowsNum);
                    // UploadFileAsync().Wait();
                    // LoadToRedshift();
                }
            }
            Console.WriteLine($"Files saved to: {outputPath}");
            File.AppendAllText(logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + " Done"));
        }
        private static void SetBucket()
        {
            OdbcCommand get_var_value = new System.Data.Odbc.OdbcCommand("SELECT var_val FROM ods.etl_var_tbl WHERE var_nm = ? ;");
            OdbcParameter param = new System.Data.Odbc.OdbcParameter("var_nm", System.Data.Odbc.OdbcType.VarChar, 256, "var_nm");
            param.Direction = System.Data.ParameterDirection.Input;
            get_var_value.Parameters.Add(param);

            using (OdbcConnection connDest = new OdbcConnection(RedshiftConnStr))
            {
                connDest.Open();
                get_var_value.Connection = connDest;
                get_var_value.Parameters["var_nm"].Value = "bucket";
                bucketName = get_var_value.ExecuteScalar() as string;
            }
        }
    }

    public class SqlToParquetExporter
    {
        private readonly string _connectionString;
        static string sqlDropDest = "";
        static string crtDestTbl = "";
        string selStr = "";
        Dictionary<string, int> scales = new Dictionary<string, int>();
        Dictionary<string, int> precisions = new Dictionary<string, int>();

        public SqlToParquetExporter(string connectionString)
        {
            _connectionString = connectionString;
        }

        public void ExportTables(string tblName, string outputPath)
        {
            string keyName = string.Empty;
            Stopwatch stopwatch = new Stopwatch();
            // Ensure output directory exists and is accessible
            EnsureDirectoryExists(outputPath);
            sqlDropDest = String.Format("DROP TABLE IF EXISTS {0}.{1}", SqlToRedshift.Program.schemaNameAWS, tblName);

            //Console.WriteLine(DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $" Exporting {tblName}...");
            //File.AppendAllText(SqlToRedshift.Program.logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $"Exporting {tblName}..." + Environment.NewLine));

            string getSelect = String.Format(@"SET NOCOUNT ON; 
            WITH CTE (Clm, column_id) AS
            (
            SELECT CASE
            WHEN DATA_TYPE IN('date', 'datetime', 'datetimeoffset') THEN 'CAST([' + COLUMN_NAME + '] AS DATETIME2) AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE IN('varchar', 'nvarchar') AND (CHARACTER_MAXIMUM_LENGTH = -1 OR CHARACTER_MAXIMUM_LENGTH > 65535) THEN 'TRANSLATE(LEFT((((CAST(ISNULL([' + COLUMN_NAME + '],'''') AS VARCHAR(MAX))) COLLATE Cyrillic_General_CI_AI) COLLATE SQL_Latin1_General_Cp850_CI_AI),65535),CHAR(173) + CHAR(160) + ''§·©®¤«»¦°±'',''- S*CR """" o~'') AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE IN('varchar', 'nvarchar') AND CHARACTER_MAXIMUM_LENGTH <> -1 THEN 'TRANSLATE(((CAST(ISNULL([' + COLUMN_NAME + '],'''') AS VARCHAR(' + CAST(CHARACTER_MAXIMUM_LENGTH AS VARCHAR) + ')) COLLATE Cyrillic_General_CI_AI) COLLATE SQL_Latin1_General_Cp850_CI_AI),CHAR(173) + CHAR(160) + ''§·©®¤«»¦°±'',''- S*CR """" o~'') AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE IN('char', 'nchar') THEN '(CAST(ISNULL([' + COLUMN_NAME + '],'''') AS CHAR(' + CAST(CHARACTER_MAXIMUM_LENGTH AS VARCHAR) + ')) COLLATE Cyrillic_General_CI_AI) COLLATE SQL_Latin1_General_Cp850_CI_AI AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE IN('tinyint') THEN 'CAST([' + COLUMN_NAME + '] AS SMALLINT) AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE = 'uniqueidentifier' THEN 'CAST([' + COLUMN_NAME + '] AS CHAR(36)) AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE IN('varbinary','binary') THEN 'CAST(''binary'' AS CHAR(6)) AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE IN('timestamp','rowversion') THEN 'CAST([' + COLUMN_NAME + '] AS BIGINT) AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE = 'hierarchyid' THEN 'CAST([' + COLUMN_NAME + '] AS VARCHAR(4000)) AS [' + COLUMN_NAME + ']'
            WHEN DATA_TYPE = 'xml' THEN 'TRANSLATE(LEFT((((CAST(ISNULL(CAST([' + COLUMN_NAME + '] AS NVARCHAR(MAX)),'''') AS VARCHAR(MAX))) COLLATE Cyrillic_General_CI_AI) COLLATE SQL_Latin1_General_Cp850_CI_AI),65535),CHAR(173) + CHAR(160) + ''§·©®¤«»¦°±'',''- S*CR """" o~'') AS [' + COLUMN_NAME + ']'
            ELSE '[' + COLUMN_NAME + ']' END
            , ORDINAL_POSITION
            FROM INFORMATION_SCHEMA.COLUMNS
            WHERE TABLE_SCHEMA = '{0}' AND TABLE_NAME = '{1}'
            )
            SELECT CAST('SET NOCOUNT ON; SELECT ' AS VARCHAR(MAX)), 0 AS column_id
            UNION ALL
            SELECT STRING_AGG(CAST(Clm AS VARCHAR(MAX)), ',')  WITHIN GROUP(ORDER BY column_id)  AS '---', 1
            FROM CTE
            UNION ALL
            SELECT ' FROM {0}.{1} WITH (NOLOCK);', 1025 AS column_id
            ORDER BY 2;", SqlToRedshift.Program.schemaNameSQL, tblName);
            string getDecNum = String.Format(@"SET NOCOUNT ON;
            SELECT COLUMN_NAME, CAST(NUMERIC_PRECISION AS INTEGER) AS NUMERIC_PRECISION, NUMERIC_SCALE FROM INFORMATION_SCHEMA.COLUMNS WHERE DATA_TYPE IN ('numeric', 'decimal') AND TABLE_SCHEMA = '{0}' AND  TABLE_NAME = '{1}'", SqlToRedshift.Program.schemaNameSQL, tblName);
            string getSQL = String.Format(@"SET NOCOUNT ON;
SELECT 'CREATE TABLE {0}.{2} (' + STRING_AGG(CAST(LOWER(QUOTENAME(c.COLUMN_NAME,CHAR(34))) as VARCHAR(MAX)) + ' ' +
CASE(c.DATA_TYPE)
WHEN 'bit' THEN 'BOOLEAN'
WHEN 'date' THEN 'DATE' 
WHEN 'time' THEN 'TIME' 
WHEN 'datetime' THEN 'TIMESTAMP' 
WHEN 'datetime2' THEN 'TIMESTAMP' 
WHEN 'smalldatetime' THEN 'TIMESTAMP' 
WHEN 'float' THEN 'FLOAT'
WHEN 'nchar' THEN 'CHAR(' + CAST(c.CHARACTER_MAXIMUM_LENGTH AS VARCHAR) + ')' 
WHEN 'char' THEN 'CHAR(' + CAST(c.CHARACTER_MAXIMUM_LENGTH AS VARCHAR) + ')' 
WHEN 'decimal' THEN ('DECIMAL(' + CAST((COALESCE(c.NUMERIC_PRECISION, 0)) AS VARCHAR) + ',' + CAST((COALESCE(c.NUMERIC_SCALE, 0)) AS VARCHAR) + ')') 
WHEN 'numeric' THEN ('NUMERIC(' + CAST((COALESCE(c.NUMERIC_PRECISION, 0)) AS VARCHAR) + ',' + CAST((COALESCE(c.NUMERIC_SCALE, 0)) AS VARCHAR) + ')') 
/* WHEN 'decimal' THEN ('DECIMAL(29,' + CAST((COALESCE(c.NUMERIC_SCALE, 0)) AS VARCHAR) + ')') 
WHEN 'numeric' THEN ('NUMERIC(29,' + CAST((COALESCE(c.NUMERIC_SCALE, 0)) AS VARCHAR) + ')') */
WHEN 'nvarchar' THEN 'VARCHAR(' + CASE WHEN c.CHARACTER_MAXIMUM_LENGTH = -1 THEN '65535' ELSE CAST(c.CHARACTER_MAXIMUM_LENGTH AS VARCHAR) END + ')' 
WHEN 'varchar' THEN 'VARCHAR(' + CASE WHEN c.CHARACTER_MAXIMUM_LENGTH = -1 THEN '65535' ELSE CAST(c.CHARACTER_MAXIMUM_LENGTH AS VARCHAR) END + ')' 
WHEN 'real' THEN 'REAL' 
WHEN 'tinyint' THEN 'SMALLINT' 
WHEN 'smallint' THEN 'SMALLINT' 
WHEN 'int' THEN 'INT'
WHEN 'bigint' THEN 'INT8'
WHEN 'uniqueidentifier' THEN 'CHAR(36)' 
WHEN 'xml' THEN 'VARCHAR(65535)' 
WHEN 'varbinary' THEN 'CHAR(6)'
WHEN 'binary' THEN 'CHAR(6)'
WHEN 'timestamp' THEN 'BIGINT'
WHEN 'rowversion' THEN 'BIGINT'
WHEN 'hierarchyid' THEN 'VARCHAR(4000)'
ELSE c.DATA_TYPE END + 
CASE WHEN c.IS_NULLABLE = 'NO' THEN ' NOT NULL ' ELSE '' END 
,',') WITHIN GROUP (ORDER BY c.ORDINAL_POSITION) + ')'
FROM INFORMATION_SCHEMA.COLUMNS c
WHERE c.TABLE_SCHEMA = '{1}' AND c.TABLE_NAME ='{2}';", SqlToRedshift.Program.schemaNameAWS, SqlToRedshift.Program.schemaNameSQL, tblName);
            stopwatch.Start(); // Start the stopwatch
            try
            {
                using (OleDbConnection connection = new OleDbConnection(_connectionString))
                {
                    OleDbCommand command = connection.CreateCommand();
                    connection.Open();
                    // get CREATE TABLE
                    command.CommandText = getSQL;
                    using (OleDbDataReader reader = command.ExecuteReader(CommandBehavior.SequentialAccess))
                    {
                        while (reader.Read())
                        {
                            crtDestTbl = reader[0].ToString();
                        }
                    }
                    command.CommandText = getDecNum;
                    precisions.Clear();
                    scales.Clear();
                    using (OleDbDataReader reader = command.ExecuteReader())
                    {
                        while (reader.Read())
                        {
                            precisions.Add(reader[0].ToString(), reader.GetInt32(1));
                            scales.Add(reader[0].ToString(), reader.GetInt32(2));
                        }
                    }

                    command.CommandText = getSelect;
                    //File.AppendAllText(SqlToRedshift.Program.logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + getSelect + Environment.NewLine));
                    using (OleDbDataReader reader = command.ExecuteReader(CommandBehavior.SequentialAccess))
                    {
                        selStr = String.Empty;
                        while (reader.Read())
                        {
                            selStr += reader[0].ToString();
                        }
                    }
                    command.CommandText = selStr;
                    File.AppendAllText(SqlToRedshift.Program.logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + selStr + Environment.NewLine));
                    var filePath = Path.Combine(outputPath, $"{tblName}.parquet");
                    using (var reader = command.ExecuteReader(CommandBehavior.SequentialAccess))
                    {
                        ExportWithBatching(reader, tblName, filePath);
                    }
                    stopwatch.Stop(); // Stop the stopwatch
                    TimeSpan elapsed = stopwatch.Elapsed;

                    Console.WriteLine(DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $" Exported {tblName} to {filePath} {elapsed.TotalSeconds} seconds");
                    File.AppendAllText(Program.logFileName, DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $" Exported {tblName} to {filePath} {elapsed.TotalSeconds} seconds" + Environment.NewLine);

                    var s3Client = new AmazonS3Client(RegionEndpoint.USEast1);
                    var fileTransferUtility = new TransferUtility(s3Client);
                    keyName = @"SQLTOAWS/" + Path.GetFileName(filePath);
                    stopwatch.Start(); // Start the stopwatch
                    try
                    {
                        // No async/await → block synchronously
                        fileTransferUtility
                            .UploadAsync(filePath, SqlToRedshift.Program.bucketName, keyName)
                            .GetAwaiter()
                            .GetResult();

                        Console.WriteLine($"Upload to {Program.bucketName} {keyName} completed successfully.");
                        File.Delete(filePath);
                    }
                    catch (AmazonS3Exception s3Ex)
                    {
                        Console.WriteLine("S3 ERROR: " + s3Ex.Message);
                        File.AppendAllText(Program.logFileName, DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + s3Ex.Message);
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine("Error uploading file: " + ex.Message);
                        File.AppendAllText(Program.logFileName, DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + ex.Message);
                    }
                    Console.WriteLine(DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + " Start Redshift COPY command");
                    using (OdbcConnection connDest = new OdbcConnection(Program.RedshiftConnStr))
                    {
                        Console.WriteLine($"Drop and create {tblName}");
                        connDest.Open();
                        sqlDropDest = String.Format("DROP TABLE IF EXISTS {0}.{1}", Program.schemaNameAWS, tblName);
                        OdbcCommand commdest = new System.Data.Odbc.OdbcCommand(sqlDropDest, connDest);
                        commdest.ExecuteNonQuery(); // drop table
                        commdest.CommandText = crtDestTbl;
                        commdest.ExecuteNonQuery(); // create table
                        string copyCommand = String.Format(@"COPY {4}.{0} FROM 's3://{1}/{2}' IAM_ROLE '{3}' PARQUET ", tblName, Program.bucketName, keyName, Program.IAMRole, Program.schemaNameAWS);
                        commdest.CommandText = copyCommand;
                        Console.WriteLine(copyCommand);
                        commdest.ExecuteNonQuery();
                        commdest.CommandText = "SELECT CAST(PG_LAST_COPY_COUNT() AS TEXT) AS pg_last_copy_count";
                        string rowsNumsStr = (commdest.ExecuteScalar()).ToString();
                        stopwatch.Stop(); // Stop the stopwatch
                        elapsed = stopwatch.Elapsed;
                        Console.WriteLine(DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $" COPY command inserted {rowsNumsStr} rows {elapsed.TotalSeconds} seconds");
                        File.AppendAllText(Program.logFileName, DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $" COPY command inserted {rowsNumsStr} rows {elapsed.TotalSeconds} seconds" + Environment.NewLine);
                    } // OdbcConnection
                    s3Client.DeleteObjectAsync(Program.bucketName, keyName);
                }
            }
            catch (Exception ex)
            {
                Console.WriteLine($"✗ Failed to export {tblName}: {ex.Message}");
            }
        }

        private void EnsureDirectoryExists(string path)
        {
            try
            {
                if (!Directory.Exists(path))
                {
                    Directory.CreateDirectory(path);
                    Console.WriteLine($"Created directory: {path}");
                }

                // Verify the directory is writable
                var testFile = Path.Combine(path, "write_test.tmp");
                File.WriteAllText(testFile, "test");
                File.Delete(testFile);
            }
            catch (Exception ex)
            {
                throw new InvalidOperationException($"Cannot access or create directory '{path}': {ex.Message}", ex);
            }
        }

        private void ExportWithBatching(OleDbDataReader reader, string tableName, string filePath)
        {
            var fieldCount = reader.FieldCount;
            var fieldTypes = new Type[fieldCount];
            var fieldNames = new string[fieldCount];


            for (int i = 0; i < fieldCount; i++)
            {
                fieldTypes[i] = reader.GetFieldType(i);
                fieldNames[i] = reader.GetName(i);
                //Console.WriteLine($"  Column {i}: {fieldNames[i]} ({fieldTypes[i].Name})");
            }

            ParquetSharp.Column[] columns = CreateColumns(fieldNames, fieldTypes);

            using var fileWriter = new ParquetFileWriter(filePath, columns);
            using var rowGroup = fileWriter.AppendRowGroup();

            // Read all data into memory first (for batching approach)
            var allData = ReadAllData(reader, fieldCount, fieldTypes);

            // Write each column sequentially
            for (int columnIndex = 0; columnIndex < fieldCount; columnIndex++)
            {
                WriteColumnToParquet(allData[columnIndex], fieldTypes[columnIndex], rowGroup);
            }

            fileWriter.Close();
            Console.WriteLine($"✓ Completed {tableName} with {allData[0].Count} rows");
            File.AppendAllText(SqlToRedshift.Program.logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + ($" ✓ Completed {tableName} with {allData[0].Count} rows") + Environment.NewLine));
        }

        private IList[] ReadAllData(OleDbDataReader reader, int fieldCount, Type[] fieldTypes)
        {
            var allData = InitializeBatchData(fieldTypes);
            var rowCount = 0;

            while (reader.Read())
            {
                for (int i = 0; i < fieldCount; i++)
                {
                    AddValueToBatch(allData[i], reader, i, fieldTypes[i]);
                }
                rowCount++;
            }

            Console.WriteLine($"  Total rows read: {rowCount}");
            File.AppendAllText(SqlToRedshift.Program.logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + $" Total rows read: {rowCount}" + Environment.NewLine));

            return allData;
        }

        private void WriteColumnToParquet(IList columnData, Type fieldType, RowGroupWriter rowGroup)
        {
            using var columnWriter = CreateColumnWriter(fieldType, rowGroup);

            if (columnData.Count == 0) return;

            switch (fieldType.Name)
            {
                case "Int32":
                    ((LogicalColumnWriter<int?>)columnWriter).WriteBatch(((List<int?>)columnData).ToArray());
                    break;
                case "Int64":
                    ((LogicalColumnWriter<long?>)columnWriter).WriteBatch(((List<long?>)columnData).ToArray());
                    break;
                case "String":
                    ((LogicalColumnWriter<string?>)columnWriter).WriteBatch(((List<string?>)columnData).ToArray());
                    break;
                case "Double":
                    ((LogicalColumnWriter<double?>)columnWriter).WriteBatch(((List<double?>)columnData).ToArray());
                    break;
                case "Decimal":
                    ((LogicalColumnWriter<decimal?>)columnWriter).WriteBatch(((List<decimal?>)columnData).ToArray());
                    break;
                case "Boolean":
                    ((LogicalColumnWriter<bool?>)columnWriter).WriteBatch(((List<bool?>)columnData).ToArray());
                    break;
                case "DateTime":
                    ((LogicalColumnWriter<DateTime?>)columnWriter).WriteBatch(((List<DateTime?>)columnData).ToArray());
                    break;
                case "Single":
                    ((LogicalColumnWriter<float?>)columnWriter).WriteBatch(((List<float?>)columnData).ToArray());
                    break;
                case "Int16":
                    ((LogicalColumnWriter<short?>)columnWriter).WriteBatch(((List<short?>)columnData).ToArray());
                    break;
                case "Byte":
                    ((LogicalColumnWriter<byte?>)columnWriter).WriteBatch(((List<byte?>)columnData).ToArray());
                    break;
                case "Guid":
                    ((LogicalColumnWriter<string?>)columnWriter).WriteBatch(((List<string?>)columnData).ToArray());
                    break;
                case "Byte[]":
                    ((LogicalColumnWriter<string?>)columnWriter).WriteBatch(((List<string?>)columnData).ToArray());
                    break;
                default:
                    ((LogicalColumnWriter<string?>)columnWriter).WriteBatch(((List<string>)columnData).ToArray());
                    break;
            }
        }

        private IList[] InitializeBatchData(Type[] fieldTypes)
        {
            var batchData = new IList[fieldTypes.Length];

            for (int i = 0; i < fieldTypes.Length; i++)
            {
                batchData[i] = fieldTypes[i].Name switch
                {
                    "Int32" => new List<int?>(),
                    "Int64" => new List<long?>(),
                    "String" => new List<string?>(),
                    "Double" => new List<double?>(),
                    "Decimal" => new List<decimal?>(),
                    "Boolean" => new List<bool?>(),
                    "DateTime" => new List<DateTime?>(),
                    "Single" => new List<float?>(),
                    "Int16" => new List<short?>(),
                    "Byte" => new List<byte?>(),
                    "Guid" => new List<string?>(),
                    "Byte[]" => new List<string?>(),
                    _ => new List<string?>()
                };
            }

            return batchData;
        }

        private void AddValueToBatch(IList batch, OleDbDataReader reader, int columnIndex, Type fieldType)
        {
            //if (reader.IsDBNull(columnIndex))
            //{
            //    AddNullToBatch(batch, fieldType);
            //    return;
            //}

            try
            {
                switch (fieldType.Name)
                {
                    case "Int32":
                        ((List<int?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetInt32(columnIndex));
                        break;
                    case "Int64":
                        ((List<long?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetInt64(columnIndex));
                        break;
                    case "String":
                        ((List<string?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetString(columnIndex));
                        break;
                    case "Double":
                        ((List<double?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetDouble(columnIndex));
                        break;
                    case "Decimal":
                        ((List<decimal?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetDecimal(columnIndex));
                        break;
                    case "Boolean":
                        ((List<bool?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetBoolean(columnIndex));
                        break;
                    case "DateTime":
                        ((List<DateTime?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetDateTime(columnIndex));
                        break;
                    case "Single":
                        ((List<float?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetFloat(columnIndex));
                        break;
                    case "Int16":
                        ((List<short?>)batch).Add(reader.GetInt16(columnIndex));
                        break;
                    case "Byte":
                        ((List<byte?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetByte(columnIndex));
                        break;
                    case "Guid":
                        ((List<string?>)batch).Add(reader.IsDBNull(columnIndex) ? null : reader.GetGuid(columnIndex).ToString());
                        break;
                    case "Byte[]":
                        var bytes = (byte[])reader.GetValue(columnIndex);
                        ((List<string>)batch).Add(Convert.ToBase64String(bytes));
                        break;
                    default:
                        var value = reader.GetValue(columnIndex);
                        ((List<string?>)batch).Add(value?.ToString() ?? string.Empty);
                        break;
                }
            }
            catch (Exception ex)
            {
                // Fallback to string conversion if specific type conversion fails
                var value = reader.GetValue(columnIndex);
                if (batch is List<string> stringList)
                {
                    stringList.Add(value?.ToString() ?? string.Empty);
                }
                else
                {
                    // If the batch list is not string type, we need to handle this differently
                    Console.WriteLine($"Warning: Type conversion failed for column {columnIndex}, using default value");
                    AddNullToBatch(batch, fieldType);
                }
                File.AppendAllText(SqlToRedshift.Program.logFileName, (DateTime.Now.ToString("MM/dd/yyyy HH:mm:ss") + ex.Message));
            }
        }

        private void AddNullToBatch(IList batch, Type fieldType)
        {
            switch (fieldType.Name)
            {
                case "Int32":
                    ((List<int?>)batch).Add(default);
                    break;
                case "Int64":
                    ((List<long?>)batch).Add(default);
                    break;
                case "String":
                    ((List<string?>)batch).Add(null);//(string.Empty);
                    break;
                case "Double":
                    ((List<double?>)batch).Add(default);
                    break;
                case "Decimal":
                    ((List<decimal?>)batch).Add(default);
                    break;
                case "Boolean":
                    ((List<bool?>)batch).Add(default);
                    break;
                case "DateTime":
                    ((List<DateTime?>)batch).Add(default);
                    break;
                case "Single":
                    ((List<float?>)batch).Add(default);
                    break;
                case "Int16":
                    ((List<short?>)batch).Add(default);
                    break;
                case "Byte":
                    ((List<byte?>)batch).Add(default);
                    break;
                case "Guid":
                    ((List<string?>)batch).Add(default);// (string.Empty);
                    break;
                case "Byte[]":
                    ((List<string>)batch).Add(string.Empty);
                    break;
                default:
                    ((List<string>)batch).Add(string.Empty);
                    break;
            }
        }

        private ParquetSharp.Column[] CreateColumns(string[] fieldNames, Type[] fieldTypes)
        {
            var columns = new ParquetSharp.Column[fieldNames.Length];
            for (int i = 0; i < fieldNames.Length; i++)
            {
                columns[i] = CreateColumn(fieldNames[i], fieldTypes[i], i);
            }
            return columns;
        }

        private ParquetSharp.Column CreateColumn(string fieldName, Type fieldType, int i)
        {
            return fieldType.Name switch
            {
                "Int32" => new Column<int?>(fieldName),
                "Int64" => new Column<long?>(fieldName),
                "String" => new Column<string?>(fieldName),
                "Double" => new Column<double?>(fieldName),
                "Decimal" => new ParquetSharp.Column(typeof(decimal?), fieldName, LogicalType.Decimal(precisions[fieldName], scales[fieldName])),
                "Boolean" => new Column<bool?>(fieldName),
                "DateTime" => new Column<DateTime?>(fieldName),
                "Single" => new Column<float?>(fieldName),
                "Int16" => new Column<short?>(fieldName),
                "Byte" => new Column<byte?>(fieldName),
                "Guid" => new Column<string?>(fieldName),
                "Byte[]" => new Column<string?>(fieldName),
                _ => new Column<string>(fieldName)
            };
        }

        private IDisposable CreateColumnWriter(Type fieldType, RowGroupWriter rowGroup)
        {
            return fieldType.Name switch
            {
                "Int32" => rowGroup.NextColumn().LogicalWriter<int?>(),
                "Int64" => rowGroup.NextColumn().LogicalWriter<long?>(),
                "String" => rowGroup.NextColumn().LogicalWriter<string?>(),
                "Double" => rowGroup.NextColumn().LogicalWriter<double?>(),
                "Decimal" => rowGroup.NextColumn().LogicalWriter<decimal?>(),
                "Boolean" => rowGroup.NextColumn().LogicalWriter<bool?>(),
                "DateTime" => rowGroup.NextColumn().LogicalWriter<DateTime?>(),
                "Single" => rowGroup.NextColumn().LogicalWriter<float?>(),
                "Int16" => rowGroup.NextColumn().LogicalWriter<short?>(),
                "Byte" => rowGroup.NextColumn().LogicalWriter<byte?>(),
                "Guid" => rowGroup.NextColumn().LogicalWriter<string?>(),
                "Byte[]" => rowGroup.NextColumn().LogicalWriter<string?>(),
                _ => rowGroup.NextColumn().LogicalWriter<string?>()
            };
        }
    }
}