unit Myc.Data.Scalar;

interface

uses
    System.SysUtils,
    Myc.Data.Decimal,
    Myc.Data.Series;

type
    // POD: Plain old data (...that fits into 64 bits)

    TScalarKind = (
        skInteger,
        skInt64,
        skUInt64,
        skSingle,
        skDouble,
        skDateTime,
        skTimestamp,
        skBoolean,
        skChar,
        skPChar,
        skString,
        skBytes,
        skDecimal
    );

    TBinaryOperator = (boAdd, boSubtract, boMultiply, boDivide, boEqual, boNotEqual, boLess, boGreater, boLessOrEqual, boGreaterOrEqual);
    TUnaryOperator = (uoNegate, uoNot);

    TScalarBytes = array[0..7] of Byte;
    TScalarPChar = array[0..3] of Char;
    TScalarString = String[7];

    TScalarValue = record
    public
        // Factory methods for creating a TScalarValue without a kind.
        class function FromInteger(AValue: Integer): TScalarValue; static; inline;
        class function FromInt64(AValue: Int64): TScalarValue; static; inline;
        class function FromUInt64(AValue: UInt64): TScalarValue; static; inline;
        class function FromSingle(AValue: Single): TScalarValue; static; inline;
        class function FromDouble(AValue: Double): TScalarValue; static; inline;
        class function FromDateTime(AValue: TDateTime): TScalarValue; static; inline;
        class function FromTimestamp(AValue: TTimestamp): TScalarValue; static; inline;
        class function FromBoolean(AValue: Boolean): TScalarValue; static; inline;
        class function FromDecimal(AValue: TDecimal): TScalarValue; static; inline;
        class function FromChar(AValue: Char): TScalarValue; static; inline;
        class function FromPChar(const AValue: String): TScalarValue; overload; static; inline;
        class function FromPChar(const AValue: TScalarPChar): TScalarValue; overload; static; inline;
        class function FromString(const AValue: String): TScalarValue; overload; static; inline;
        class function FromString(const AValue: TScalarString): TScalarValue; overload; static; inline;
        class function FromBytes(const AValue: TScalarBytes): TScalarValue; static; inline;

    // Direct field access for performance.
    case TScalarKind of
        skInteger: (AsInteger: Integer);
        skInt64: (AsInt64: Int64);
        skUInt64: (AsUInt64: UInt64);
        skSingle: (AsSingle: Single);
        skDouble: (AsDouble: Double);
        skDateTime: (AsDateTime: TDateTime);
        skTimestamp: (AsTimestamp: TTimestamp);
        skBoolean: (AsBoolean: Boolean);
        skChar: (AsChar: Char);
        skPChar: (AsPChar: TScalarPChar);
        skString: (AsString: String[7]);
        skBytes: (AsBytes: TScalarBytes);
        skDecimal: (AsDecimal: TDecimal);
    end;

    // A scalar value with a type identifier.
    TScalar = record
    public
        Kind: TScalarKind;
        Value: TScalarValue;

        // Creates a new scalar from a kind and a value.
        constructor Create(AKind: TScalarKind; const AValue: TScalarValue);

        // Factory methods for specific scalar types (do not use anymore, these may become deprecated in the future)
        class function FromInteger(AValue: Integer): TScalar; static; inline;
        class function FromInt64(AValue: Int64): TScalar; static; inline;
        class function FromUInt64(AValue: UInt64): TScalar; static; inline;
        class function FromSingle(AValue: Single): TScalar; static; inline;
        class function FromDouble(AValue: Double): TScalar; static; inline;
        class function FromDateTime(AValue: TDateTime): TScalar; static; inline;
        class function FromTimestamp(AValue: TTimestamp): TScalar; static; inline;
        class function FromBoolean(AValue: Boolean): TScalar; static; inline;
        class function FromChar(AValue: Char): TScalar; static; inline;
        class function FromPChar(const AValue: TScalarPChar): TScalar; static; inline;
        class function FromString(const AValue: String): TScalar; static; inline;
        class function FromBytes(const AValue: TScalarBytes): TScalar; static; inline;
        class function FromDecimal(const AValue: TDecimal): TScalar; static; inline;

        // Implicit casts from primitive types to TScalar.
        class operator Implicit(AValue: Integer): TScalar; overload; inline;
        class operator Implicit(AValue: Int64): TScalar; overload; inline;
        class operator Implicit(AValue: UInt64): TScalar; overload; inline;
        class operator Implicit(AValue: Single): TScalar; overload; inline;
        class operator Implicit(AValue: Double): TScalar; overload; inline;
        class operator Implicit(AValue: TDateTime): TScalar; overload; inline;
        class operator Implicit(AValue: Boolean): TScalar; overload; inline;
        class operator Implicit(AValue: Char): TScalar; overload; inline;
        class operator Implicit(const AValue: TDecimal): TScalar; overload; inline;

        class function StringToKind(const AName: string): TScalarKind; static;

        function ToString: String;

        class function IsBinaryOperatorSupported(Op: TBinaryOperator; A, B: TScalarKind): Boolean; static;
        class function IsUnaryOperatorSupported(Op: TUnaryOperator; A: TScalarKind): Boolean; static;

        class function TryBinaryOperation(Op: TBinaryOperator; const A, B: TScalar; out Res: TScalar): Boolean; static;
        class function TryUnaryOperation(Op: TUnaryOperator; const A: TScalar; out Res: TScalar): Boolean; static;

        class operator Add(const A, B: TScalar): TScalar;
        class operator Divide(const A, B: TScalar): TScalar;
        class operator Equal(const A, B: TScalar): Boolean;
        class operator GreaterThan(const A, B: TScalar): Boolean;
        class operator GreaterThanOrEqual(const A, B: TScalar): Boolean;
        class operator LessThan(const A, B: TScalar): Boolean;
        class operator LessThanOrEqual(const A, B: TScalar): Boolean;
        class operator LogicalNot(const A: TScalar): TScalar;
        class operator Multiply(const A, B: TScalar): TScalar;
        class operator Negative(const A: TScalar): TScalar;
        class operator NotEqual(const A, B: TScalar): Boolean;
        class operator Round(const A: TScalar): TScalar;
        class operator Subtract(const A, B: TScalar): TScalar;
        class operator Trunc(const A: TScalar): TScalar;
    end;

    TScalarKindHelper = record helper for TScalarKind
    public
        function ToString: string;
    end;

    // Basic data structures using the scalar type (these are not POD of course)

    // An array of scalar values of the same kind.
    TScalarArray = record
    public
        Kind: TScalarKind;
        Items: TArray<TScalarValue>;

        // Creates a new scalar array.
        constructor Create(AKind: TScalarKind; const AItems: TArray<TScalarValue>);
    end;

    TScalarTuple = TArray<TScalar>;

    // A field definition for a scalar record.
    TScalarRecordField = record
        Name: String;
        Kind: TScalarKind;
        constructor Create(const AName: String; AKind: TScalarKind);
    end;

    TScalarRecordDefinition = record
    private
        FFields: TArray<TScalarRecordField>;
    public
        constructor Create(const AFields: TArray<TScalarRecordField>);
        function IndexOf(const Name: String): Integer;
        property Fields: TArray<TScalarRecordField> read FFields;
    end;

    // A record of scalar values, based on a definition.
    TScalarRecord = record
    public
        // Creates a new scalar record.
        constructor Create(const ADef: TScalarRecordDefinition; const AFields: TArray<TScalarValue>);

        function GetDef: TScalarRecordDefinition; inline;
        function GetFields: TArray<TScalarValue>; inline;
        function GetItems(const Name: String): TScalar;

        property Def: TScalarRecordDefinition read GetDef;
        property Fields: TArray<TScalarValue> read GetFields;
        property Items[const Name: String]: TScalar read GetItems; default;

    strict private
        FDef: TScalarRecordDefinition;
        FFields: TArray<TScalarValue>;
    end;

    ISeries = interface
        {$region 'private'}
        function GetCount: Int64;
        function GetItems(Idx: Integer): TScalar;
        function GetTotalCount: Int64;
        {$endregion}
        property Count: Int64 read GetCount;
        property Items[Idx: Integer]: TScalar read GetItems; default;
        property TotalCount: Int64 read GetTotalCount;
    end;

    IWriteableSeries = interface(ISeries)
        procedure Add(const Item: TScalarValue; Lookback: Int64 = -1);
    end;

    // A time series of scalar records, optimized for memory and access speed.
    TScalarSeries = class(TInterfacedObject, ISeries, IWriteableSeries)
    private
        FKind: TScalarKind;
        FArray: TChunkArray<TScalarValue>;
        FTotalCount: Int64;

        function GetCount: Int64; inline;
        function GetItems(Idx: Integer): TScalar; inline;
        function GetTotalCount: Int64; inline;

    public
        constructor Create(AKind: TScalarKind);
        procedure Add(const Item: TScalarValue; Lookback: Int64 = -1);
    end;

    IRecordSeries = interface
        {$region 'private'}
        function GetCount: Int64;
        function GetDef: TScalarRecordDefinition;
        function GetItems(Idx: Integer): TScalarRecord;
        function GetTotalCount: Int64;
        {$endregion}
        procedure Add(const Item: TScalarRecord; Lookback: Int64 = -1);
        function CreateMemberSeries(const Field: String): ISeries;
        property Count: Int64 read GetCount;
        property Def: TScalarRecordDefinition read GetDef;
        property Items[Idx: Integer]: TScalarRecord read GetItems; default;
        property TotalCount: Int64 read GetTotalCount;
    end;

    // A time series of scalar records, optimized for memory and access speed.
    TScalarRecordSeries = class(TInterfacedObject, IRecordSeries)
        type
            TMemberSeries = class(TInterfacedObject, ISeries)
            private
                FRecordSeries: TScalarRecordSeries;
                FKind: TScalarKind;
                FOffset: Integer;
                function GetCount: Int64;
                function GetItems(Idx: Integer): TScalar;
                function GetTotalCount: Int64;
            public
                constructor Create(ARecordSeries: TScalarRecordSeries; AElementIdx: Integer);
                destructor Destroy; override;
            end;

    private
        FDef: TScalarRecordDefinition;
        FArray: TChunkArray<TScalarValue>;
        FTotalCount: Int64;
        function GetCount: Int64; inline;
        function GetDef: TScalarRecordDefinition; inline;
        function GetItems(Idx: Integer): TScalarRecord; inline;
        function GetTotalCount: Int64; inline;
        function GetItemRef(Idx: Integer): TChunkArray<TScalarValue>.PT; inline;
    public
        constructor Create(const ADef: TScalarRecordDefinition);
        procedure Add(const Item: TScalarRecord; Lookback: Int64 = -1);
        function CreateMemberSeries(const Field: String): ISeries;
        property Count: Int64 read GetCount;
        property Def: TScalarRecordDefinition read GetDef;
        property Items[Idx: Integer]: TScalarRecord read GetItems; default;
        property TotalCount: Int64 read GetTotalCount;
    end;

    TIndexSeries = class(TInterfacedObject, ISeries)
    private
        FIndexArray: TArray<Integer>;
        function GetCount: Int64;
        function GetTotalCount: Int64;
        function GetItems(Idx: Integer): TScalar;
    public
        constructor Create(const AIndexArray: TArray<Integer>);
    end;

    TBinaryOperatorHelper = record helper for TBinaryOperator
        function ToString: string;
    end;

    TUnaryOperatorHelper = record helper for TUnaryOperator
        function ToString: string;
    end;

implementation

{ TScalarValue }

class function TScalarValue.FromBoolean(AValue: Boolean): TScalarValue;
begin
    Result.AsBoolean := AValue;
end;

class function TScalarValue.FromBytes(const AValue: TScalarBytes): TScalarValue;
begin
    Result.AsBytes := AValue;
end;

class function TScalarValue.FromChar(AValue: Char): TScalarValue;
begin
    Result.AsChar := AValue;
end;

class function TScalarValue.FromDateTime(AValue: TDateTime): TScalarValue;
begin
    Result.AsDateTime := AValue;
end;

class function TScalarValue.FromDecimal(AValue: TDecimal): TScalarValue;
begin
    Result.AsDecimal := AValue;
end;

class function TScalarValue.FromDouble(AValue: Double): TScalarValue;
begin
    Result.AsDouble := AValue;
end;

class function TScalarValue.FromInt64(AValue: Int64): TScalarValue;
begin
    Result.AsInt64 := AValue;
end;

class function TScalarValue.FromInteger(AValue: Integer): TScalarValue;
begin
    Result.AsInteger := AValue;
end;

class function TScalarValue.FromPChar(const AValue: String): TScalarValue;
var
    len, i: Integer;
begin
    FillChar(Result.AsPChar, SizeOf(Result.AsPChar), #0);
    len := Length(AValue);
    if (len > 0) then
    begin
        if (len > Length(Result.AsPChar)) then
            len := Length(Result.AsPChar);
        for i := 0 to len - 1 do
            Result.AsPChar[i] := AValue[i + 1];
    end;
end;

class function TScalarValue.FromPChar(const AValue: TScalarPChar): TScalarValue;
begin
    Result.AsPChar := AValue;
end;

class function TScalarValue.FromSingle(AValue: Single): TScalarValue;
begin
    Result.AsSingle := AValue;
end;

class function TScalarValue.FromString(const AValue: String): TScalarValue;
begin
    Result.AsString := ShortString(AValue);
end;

class function TScalarValue.FromString(const AValue: TScalarString): TScalarValue;
begin
    Result.AsString := AValue;
end;

class function TScalarValue.FromTimestamp(AValue: TTimestamp): TScalarValue;
begin
    Result.AsTimestamp := AValue;
end;

class function TScalarValue.FromUInt64(AValue: UInt64): TScalarValue;
begin
    Result.AsUInt64 := AValue;
end;

{ TScalar }

constructor TScalar.Create(AKind: TScalarKind; const AValue: TScalarValue);
begin
    Kind := AKind;
    Value := AValue;
end;

class function TScalar.FromBoolean(AValue: Boolean): TScalar;
begin
    Result.Kind := skBoolean;
    Result.Value := TScalarValue.FromBoolean(AValue);
end;

class function TScalar.FromBytes(const AValue: TScalarBytes): TScalar;
begin
    Result.Kind := skBytes;
    Result.Value := TScalarValue.FromBytes(AValue);
end;

class function TScalar.FromChar(AValue: Char): TScalar;
begin
    Result.Kind := skChar;
    Result.Value := TScalarValue.FromChar(AValue);
end;

class function TScalar.FromDateTime(AValue: TDateTime): TScalar;
begin
    Result.Kind := skDateTime;
    Result.Value := TScalarValue.FromDateTime(AValue);
end;

class function TScalar.FromDecimal(const AValue: TDecimal): TScalar;
begin
    Result.Kind := skDecimal;
    Result.Value := TScalarValue.FromDecimal(AValue);
end;

class function TScalar.FromDouble(AValue: Double): TScalar;
begin
    Result.Kind := skDouble;
    Result.Value := TScalarValue.FromDouble(AValue);
end;

class operator TScalar.Implicit(AValue: Boolean): TScalar;
begin
    Result.Kind := skBoolean;
    Result.Value := TScalarValue.FromBoolean(AValue);
end;

class operator TScalar.Implicit(AValue: Char): TScalar;
begin
    Result.Kind := skChar;
    Result.Value := TScalarValue.FromChar(AValue);
end;

class operator TScalar.Implicit(AValue: TDateTime): TScalar;
begin
    Result.Kind := skDateTime;
    Result.Value := TScalarValue.FromDateTime(AValue);
end;

class operator TScalar.Implicit(const AValue: TDecimal): TScalar;
begin
    Result.Kind := skDecimal;
    Result.Value := TScalarValue.FromDecimal(AValue);
end;

class operator TScalar.Implicit(AValue: Double): TScalar;
begin
    Result.Kind := skDouble;
    Result.Value := TScalarValue.FromDouble(AValue);
end;

class operator TScalar.Implicit(AValue: Int64): TScalar;
begin
    Result.Kind := skInt64;
    Result.Value := TScalarValue.FromInt64(AValue);
end;

class operator TScalar.Implicit(AValue: Integer): TScalar;
begin
    Result.Kind := skInteger;
    Result.Value := TScalarValue.FromInteger(AValue);
end;

class operator TScalar.Implicit(AValue: Single): TScalar;
begin
    Result.Kind := skSingle;
    Result.Value := TScalarValue.FromSingle(AValue);
end;

class operator TScalar.Implicit(AValue: UInt64): TScalar;
begin
    Result.Kind := skUInt64;
    Result.Value := TScalarValue.FromUInt64(AValue);
end;

class function TScalar.FromInt64(AValue: Int64): TScalar;
begin
    Result.Kind := skInt64;
    Result.Value := TScalarValue.FromInt64(AValue);
end;

class function TScalar.FromInteger(AValue: Integer): TScalar;
begin
    Result.Kind := skInteger;
    Result.Value := TScalarValue.FromInteger(AValue);
end;

class function TScalar.FromPChar(const AValue: TScalarPChar): TScalar;
begin
    Result.Kind := skPChar;
    Result.Value := TScalarValue.FromPChar(AValue);
end;

class function TScalar.FromSingle(AValue: Single): TScalar;
begin
    Result.Kind := skSingle;
    Result.Value := TScalarValue.FromSingle(AValue);
end;

class function TScalar.FromString(const AValue: String): TScalar;
begin
    Result.Kind := skString;
    Result.Value := TScalarValue.FromString(AValue);
end;

class function TScalar.FromTimestamp(AValue: TTimestamp): TScalar;
begin
    Result.Kind := skTimestamp;
    Result.Value := TScalarValue.FromTimestamp(AValue);
end;

class function TScalar.FromUInt64(AValue: UInt64): TScalar;
begin
    Result.Kind := skUInt64;
    Result.Value := TScalarValue.FromUInt64(AValue);
end;

class function TScalar.IsBinaryOperatorSupported(Op: TBinaryOperator; A, B: TScalarKind): Boolean;
begin
    const FLOAT_KINDS: set of TScalarKind = [skSingle, skDouble];
    const ORDINAL_KINDS: set of TScalarKind = [skInteger, skInt64, skUInt64];
    const NUMERIC_KINDS: set of TScalarKind = FLOAT_KINDS + ORDINAL_KINDS + [skDecimal];
    const COMPARABLE_KINDS: set of TScalarKind = NUMERIC_KINDS + [skDateTime, skChar, skString];

    // General check for incompatible types
    if (A = B) then
    begin
        case Op of
            boAdd, boSubtract, boMultiply, boDivide: Result := A in NUMERIC_KINDS + [skDateTime];
            boEqual, boNotEqual, boLess, boGreater, boLessOrEqual, boGreaterOrEqual: Result := A in COMPARABLE_KINDS;
        else
            Result := False;
        end;
    end
    else
    begin
        // Mixed-type checks
        case Op of
            boAdd:
                Result :=
                    (((A in NUMERIC_KINDS) and (B in NUMERIC_KINDS))
                        or ((A = skDateTime) and (B in ORDINAL_KINDS + FLOAT_KINDS))
                        or ((B = skDateTime) and (A in ORDINAL_KINDS + FLOAT_KINDS)));
            boSubtract:
                Result := (((A in NUMERIC_KINDS) and (B in NUMERIC_KINDS)) or ((A = skDateTime) and (B in NUMERIC_KINDS + [skDateTime])));
            boMultiply, boDivide: Result := (A in NUMERIC_KINDS) and (B in NUMERIC_KINDS);
            boEqual, boNotEqual, boLess, boGreater, boLessOrEqual, boGreaterOrEqual:
                Result := (A in NUMERIC_KINDS) and (B in NUMERIC_KINDS);
        else
            Result := False;
        end;
    end;

    // Specific exclusion rules
    if Result then
    begin
        // Decimal cannot be mixed with floats for any operator
        if ((A = skDecimal) and (B in FLOAT_KINDS)) or ((B = skDecimal) and (A in FLOAT_KINDS)) then
            Result :=
                False
                    // Cannot subtract a DateTime from a Numeric type
        else if (Op = boSubtract) and (A in NUMERIC_KINDS) and (B = skDateTime) then
            Result := False;
    end;
end;

class function TScalar.IsUnaryOperatorSupported(Op: TUnaryOperator; A: TScalarKind): Boolean;
begin
    case Op of
        uoNegate: Result := A in [skInteger, skInt64, skSingle, skDouble, skDecimal];
        uoNot: Result := A in [skBoolean, skInteger, skInt64, skUInt64];
    else
        Result := False;
    end;
end;

class function TScalar.StringToKind(const AName: string): TScalarKind;
begin
    if SameText(AName, 'Integer') then
        Result := skInteger
    else if SameText(AName, 'Int64') then
        Result := skInt64
    else if SameText(AName, 'UInt64') then
        Result := skUInt64
    else if SameText(AName, 'Single') then
        Result := skSingle
    else if SameText(AName, 'Double') then
        Result := skDouble
    else if SameText(AName, 'DateTime') then
        Result := skDateTime
    else if SameText(AName, 'Timestamp') then
        Result := skTimestamp
    else if SameText(AName, 'Boolean') then
        Result := skBoolean
    else if SameText(AName, 'Char') then
        Result := skChar
    else if SameText(AName, 'PChar') then
        Result := skPChar
    else if SameText(AName, 'String') then
        Result := skString
    else if SameText(AName, 'Bytes') then
        Result := skBytes
    else if SameText(AName, 'Decimal') then
        Result := skDecimal
    else
        raise EArgumentException.CreateFmt('Unknown scalar type name: "%s"', [AName]);
end;

function TScalar.ToString: String;
begin
    case Kind of
        skInteger: Result := IntToStr(Value.AsInteger);
        skInt64: Result := IntToStr(Value.AsInt64);
        skUInt64: Result := UIntToStr(Value.AsUInt64);
        skSingle: Result := FloatToStr(Value.AsSingle);
        skDouble: Result := FloatToStr(Value.AsDouble);
        skDateTime: Result := DateTimeToStr(Value.AsDateTime);
        skTimestamp: Result := FormatDateTime('yyyy-mm-dd hh:nn:ss.zzz', TimeStampToDateTime(Value.AsTimestamp));
        skBoolean: Result := BoolToStr(Value.AsBoolean, True);
        skChar: Result := '''' + Value.AsChar + '''';
        skPChar: Result := PChar(@Value.AsPChar); // Needs explicit cast/pointer
        skString: Result := '''' + string(Value.AsString) + '''';
        skBytes: Result := '(Bytes)';
        skDecimal: Result := FloatToStr(Double(Value.AsDecimal));
    else
        Result := '[Unknown Scalar]';
    end;
end;

class function TScalar.TryBinaryOperation(Op: TBinaryOperator; const A, B: TScalar; out Res: TScalar): Boolean;
begin
    if not IsBinaryOperatorSupported(Op, A.Kind, B.Kind) then
    begin
        Result := False;
        exit;
    end;

    try
        case Op of
            boAdd: Res := A + B;
            boSubtract: Res := A - B;
            boMultiply: Res := A * B;
            boDivide: Res := A / B;
            boEqual: Res := TScalar.FromBoolean(A = B);
            boNotEqual: Res := TScalar.FromBoolean(A <> B);
            boLess: Res := TScalar.FromBoolean(A < B);
            boGreater: Res := TScalar.FromBoolean(A > B);
            boLessOrEqual: Res := TScalar.FromBoolean(A <= B);
            boGreaterOrEqual: Res := TScalar.FromBoolean(A >= B);
        else
            Result := False;
            exit;
        end;
        Result := True;
    except
        Result := False;
    end;
end;

class function TScalar.TryUnaryOperation(Op: TUnaryOperator; const A: TScalar; out Res: TScalar): Boolean;
begin
    if not IsUnaryOperatorSupported(Op, A.Kind) then
    begin
        Result := False;
        exit;
    end;

    try
        case Op of
            uoNegate: Res := -A;
            uoNot: Res := not A;
        else
            Result := False;
            exit;
        end;
        Result := True;
    except
        Result := False;
    end;
end;

class operator TScalar.Add(const A, B: TScalar): TScalar;
begin
    // Fast path for identical types
    if (A.Kind = B.Kind) then
    begin
        Result.Kind := A.Kind;
        case A.Kind of
            skInteger: Result.Value.AsInteger := A.Value.AsInteger + B.Value.AsInteger;
            skInt64: Result.Value.AsInt64 := A.Value.AsInt64 + B.Value.AsInt64;
            skUInt64: Result.Value.AsUInt64 := A.Value.AsUInt64 + B.Value.AsUInt64;
            skSingle: Result.Value.AsSingle := A.Value.AsSingle + B.Value.AsSingle;
            skDouble: Result.Value.AsDouble := A.Value.AsDouble + B.Value.AsDouble;
            skDecimal: Result.Value.AsDecimal := A.Value.AsDecimal + B.Value.AsDecimal;
        else
            raise EArgumentException.CreateFmt('Operator Add not supported for type %s', [A.Kind.ToString]);
        end;
        exit;
    end;

    // Slow path for mixed types with compatible conversion
    const FLOAT_KINDS: set of TScalarKind = [skSingle, skDouble];
    const ORDINAL_KINDS: set of TScalarKind = [skInteger, skInt64, skUInt64];
    const NUMERIC_KINDS: set of TScalarKind = FLOAT_KINDS + ORDINAL_KINDS + [skDecimal];

    if not (((A.Kind in NUMERIC_KINDS) and (B.Kind in NUMERIC_KINDS))
        or ((A.Kind = skDateTime) and (B.Kind in ORDINAL_KINDS + FLOAT_KINDS))
        or ((B.Kind = skDateTime) and (A.Kind in ORDINAL_KINDS + FLOAT_KINDS))) then
        raise EArgumentException.CreateFmt('Operator Add requires compatible types, but got %s and %s', [A.Kind.ToString, B.Kind.ToString]);

    if (A.Kind = skDecimal) or (B.Kind = skDecimal) then
    begin
        if (A.Kind in FLOAT_KINDS) or (B.Kind in FLOAT_KINDS) then
            raise EArgumentException
                .CreateFmt('Operator Add cannot mix Decimal and floating-point types (%s, %s)', [A.Kind.ToString, B.Kind.ToString]);

        var valA, valB: TDecimal;
        case A.Kind of
            skInteger: valA := TDecimal.Create(A.Value.AsInteger, 0);
            skInt64: valA := TDecimal.Create(A.Value.AsInt64, 0);
            skUInt64: valA := TDecimal.Create(Int64(A.Value.AsUInt64), 0);
            skDecimal: valA := A.Value.AsDecimal;
        end;
        case B.Kind of
            skInteger: valB := TDecimal.Create(B.Value.AsInteger, 0);
            skInt64: valB := TDecimal.Create(B.Value.AsInt64, 0);
            skUInt64: valB := TDecimal.Create(Int64(B.Value.AsUInt64), 0);
            skDecimal: valB := B.Value.AsDecimal;
        end;
        Result := TScalar.FromDecimal(valA + valB);
    end
    else if (A.Kind = skDateTime) or (B.Kind = skDateTime) then
    begin
        if A.Kind = skDateTime then
        begin
            var dtPart := A.Value.AsDateTime;
            var numPart: Double := 0;
            case B.Kind of
                skInteger: numPart := B.Value.AsInteger;
                skInt64: numPart := B.Value.AsInt64;
                skUInt64: numPart := B.Value.AsUInt64;
                skSingle: numPart := B.Value.AsSingle;
                skDouble: numPart := B.Value.AsDouble;
            end;
            Result := TScalar.FromDateTime(dtPart + numPart);
        end
        else
        begin
            var dtPart := B.Value.AsDateTime;
            var numPart: Double := 0;
            case A.Kind of
                skInteger: numPart := A.Value.AsInteger;
                skInt64: numPart := A.Value.AsInt64;
                skUInt64: numPart := A.Value.AsUInt64;
                skSingle: numPart := A.Value.AsSingle;
                skDouble: numPart := A.Value.AsDouble;
            end;
            Result := TScalar.FromDateTime(dtPart + numPart);
        end;
    end
    else if (A.Kind = skDouble) or (B.Kind = skDouble) then
    begin
        var dblA: Double := 0;
        var dblB: Double := 0;
        case A.Kind of
            skInteger: dblA := A.Value.AsInteger;
            skInt64: dblA := A.Value.AsInt64;
            skUInt64: dblA := A.Value.AsUInt64;
            skSingle: dblA := A.Value.AsSingle;
            skDouble: dblA := A.Value.AsDouble;
        end;
        case B.Kind of
            skInteger: dblB := B.Value.AsInteger;
            skInt64: dblB := B.Value.AsInt64;
            skUInt64: dblB := B.Value.AsUInt64;
            skSingle: dblB := B.Value.AsSingle;
            skDouble: dblB := B.Value.AsDouble;
        end;
        Result := TScalar.FromDouble(dblA + dblB);
    end
    else if (A.Kind = skSingle) or (B.Kind = skSingle) then
    begin
        var sngA: Single := 0;
        var sngB: Single := 0;
        case A.Kind of
            skInteger: sngA := A.Value.AsInteger;
            skInt64: sngA := A.Value.AsInt64;
            skUInt64: sngA := A.Value.AsUInt64;
            skSingle: sngA := A.Value.AsSingle;
        end;
        case B.Kind of
            skInteger: sngB := B.Value.AsInteger;
            skInt64: sngB := B.Value.AsInt64;
            skUInt64: sngB := B.Value.AsUInt64;
            skSingle: sngB := B.Value.AsSingle;
        end;
        Result := TScalar.FromSingle(sngA + sngB);
    end
    else if (A.Kind = skUInt64) or (B.Kind = skUInt64) then
    begin
        var u64A: UInt64 := 0;
        var u64B: UInt64 := 0;
        case A.Kind of
            skInteger: u64A := UInt64(A.Value.AsInteger);
            skInt64: u64A := UInt64(A.Value.AsInt64);
            skUInt64: u64A := A.Value.AsUInt64;
        end;
        case B.Kind of
            skInteger: u64B := UInt64(B.Value.AsInteger);
            skInt64: u64B := UInt64(B.Value.AsInt64);
            skUInt64: u64B := B.Value.AsUInt64;
        end;
        Result := TScalar.FromUInt64(u64A + u64B);
    end
    else if (A.Kind = skInt64) or (B.Kind = skInt64) then
    begin
        var i64A: Int64 := 0;
        var i64B: Int64 := 0;
        case A.Kind of
            skInteger: i64A := A.Value.AsInteger;
            skInt64: i64A := A.Value.AsInt64;
        end;
        case B.Kind of
            skInteger: i64B := B.Value.AsInteger;
            skInt64: i64B := B.Value.AsInt64;
        end;
        Result := TScalar.FromInt64(i64A + i64B);
    end;
end;

class operator TScalar.Divide(const A, B: TScalar): TScalar;
begin
    // Fast path for identical types
    if (A.Kind = B.Kind) then
    begin
        case A.Kind of
            skInteger: Result := TScalar.FromDouble(A.Value.AsInteger / B.Value.AsInteger);
            skInt64: Result := TScalar.FromDouble(A.Value.AsInt64 / B.Value.AsInt64);
            skUInt64: Result := TScalar.FromDouble(A.Value.AsUInt64 / B.Value.AsUInt64);
            skSingle: Result := TScalar.FromSingle(A.Value.AsSingle / B.Value.AsSingle);
            skDouble: Result := TScalar.FromDouble(A.Value.AsDouble / B.Value.AsDouble);
            skDecimal: Result := TScalar.FromDecimal(A.Value.AsDecimal / B.Value.AsDecimal);
        else
            raise EArgumentException.CreateFmt('Operator Divide not supported for type %s', [A.Kind.ToString]);
        end;
        exit;
    end;

    // Slow path for mixed types
    const FLOAT_KINDS: set of TScalarKind = [skSingle, skDouble];
    const NUMERIC_KINDS: set of TScalarKind = FLOAT_KINDS + [skInteger, skInt64, skUInt64, skDecimal];

    if not ((A.Kind in NUMERIC_KINDS) and (B.Kind in NUMERIC_KINDS)) then
        raise EArgumentException.CreateFmt('Operator Divide not supported for types %s and %s', [A.Kind.ToString, B.Kind.ToString]);

    if (A.Kind = skDecimal) or (B.Kind = skDecimal) then
    begin
        if (A.Kind in FLOAT_KINDS) or (B.Kind in FLOAT_KINDS) then
            raise EArgumentException
                .CreateFmt('Operator Divide cannot mix Decimal and floating-point types (%s, %s)', [A.Kind.ToString, B.Kind.ToString]);
        var valA, valB: TDecimal;
        case A.Kind of
            skInteger: valA := TDecimal.Create(A.Value.AsInteger, 0);
            skInt64: valA := TDecimal.Create(A.Value.AsInt64, 0);
            skUInt64: valA := TDecimal.Create(Int64(A.Value.AsUInt64), 0);
            skDecimal: valA := A.Value.AsDecimal;
        end;
        case B.Kind of
            skInteger: valB := TDecimal.Create(B.Value.AsInteger, 0);
            skInt64: valB := TDecimal.Create(B.Value.AsInt64, 0);
            skUInt64: valB := TDecimal.Create(Int64(B.Value.AsUInt64), 0);
            skDecimal: valB := B.Value.AsDecimal;
        end;
        Result := TScalar.FromDecimal(valA / valB);
    end
    else
    begin
        var dblA: Double := 0.0;
        var dblB: Double := 0.0;
        case A.Kind of
            skInteger: dblA := A.Value.AsInteger;
            skInt64: dblA := A.Value.AsInt64;
            skUInt64: dblA := A.Value.AsUInt64;
            skSingle: dblA := A.Value.AsSingle;
            skDouble: dblA := A.Value.AsDouble;
        end;
        case B.Kind of
            skInteger: dblB := B.Value.AsInteger;
            skInt64: dblB := B.Value.AsInt64;
            skUInt64: dblB := B.Value.AsUInt64;
            skSingle: dblB := B.Value.AsSingle;
            skDouble: dblB := B.Value.AsDouble;
        end;
        Result := TScalar.FromDouble(dblA / dblB);
    end;
end;

class operator TScalar.Equal(const A, B: TScalar): Boolean;
begin
    // Fast path for identical types
    if (A.Kind = B.Kind) then
    begin
        case A.Kind of
            skInteger: Result := (A.Value.AsInteger = B.Value.AsInteger);
            skInt64: Result := (A.Value.AsInt64 = B.Value.AsInt64);
            skUInt64: Result := (A.Value.AsUInt64 = B.Value.AsUInt64);
            skSingle: Result := (A.Value.AsSingle = B.Value.AsSingle);
            skDouble: Result := (A.Value.AsDouble = B.Value.AsDouble);
            skDateTime: Result := (A.Value.AsDateTime = B.Value.AsDateTime);
            skTimestamp:
                Result := (A.Value.AsTimestamp.Time = B.Value.AsTimestamp.Time) and (A.Value.AsTimestamp.Date = B.Value.AsTimestamp.Date);
            skBoolean: Result := (A.Value.AsBoolean = B.Value.AsBoolean);
            skChar: Result := (A.Value.AsChar = B.Value.AsChar);
            skPChar: Result := (StrComp(PChar(@A.Value.AsPChar), PChar(@B.Value.AsPChar)) = 0);
            skString: Result := (A.Value.AsString = B.Value.AsString);
            skBytes: Result := CompareMem(@A.Value.AsBytes, @B.Value.AsBytes, SizeOf(TScalarBytes));
            skDecimal: Result := (A.Value.AsDecimal = B.Value.AsDecimal);
        else
            Result := False;
        end;
        exit;
    end;

    // Slow path for mixed, but compatible, types
    const FLOAT_KINDS: set of TScalarKind = [skSingle, skDouble];
    const NUMERIC_KINDS: set of TScalarKind = FLOAT_KINDS + [skInteger, skInt64, skUInt64, skDecimal];

    if not ((A.Kind in NUMERIC_KINDS) and (B.Kind in NUMERIC_KINDS)) then
    begin
        Result := False;
        exit;
    end;

    if ((A.Kind = skDecimal) and (B.Kind in FLOAT_KINDS)) or ((B.Kind = skDecimal) and (A.Kind in FLOAT_KINDS)) then
    begin
        Result := False;
        exit;
    end;

    if (A.Kind = skDecimal) or (B.Kind = skDecimal) then
    begin
        var valA, valB: TDecimal;
        case A.Kind of
            skInteger: valA := TDecimal.Create(A.Value.AsInteger, 0);
            skInt64: valA := TDecimal.Create(A.Value.AsInt64, 0);
            skUInt64: valA := TDecimal.Create(Int64(A.Value.AsUInt64), 0);
            skDecimal: valA := A.Value.AsDecimal;
        else
            Result := False;
            exit;
        end;
        case B.Kind of
            skInteger: valB := TDecimal.Create(B.Value.AsInteger, 0);
            skInt64: valB := TDecimal.Create(B.Value.AsInt64, 0);
            skUInt64: valB := TDecimal.Create(Int64(B.Value.AsUInt64), 0);
            skDecimal: valB := B.Value.AsDecimal;
        else
            Result := False;
            exit;
        end;
        Result := (valA = valB);
    end
    else
    begin
        var dblA: Double := 0.0;
        var dblB: Double := 0.0;
        case A.Kind of
            skInteger: dblA := A.Value.AsInteger;
            skInt64: dblA := A.Value.AsInt64;
            skUInt64: dblA := A.Value.AsUInt64;
            skSingle: dblA := A.Value.AsSingle;
            skDouble: dblA := A.Value.AsDouble;
        end;
        case B.Kind of
            skInteger: dblB := B.Value.AsInteger;
            skInt64: dblB := B.Value.AsInt64;
            skUInt64: dblB := B.Value.AsUInt64;
            skSingle: dblB := B.Value.AsSingle;
            skDouble: dblB := B.Value.AsDouble;
        end;
        Result := (dblA = dblB);
    end;
end;

class operator TScalar.GreaterThan(const A, B: TScalar): Boolean;
begin
    // Fast path for identical types
    if (A.Kind = B.Kind) then
    begin
        case A.Kind of
            skInteger: Result := (A.Value.AsInteger > B.Value.AsInteger);
            skInt64: Result := (A.Value.AsInt64 > B.Value.AsInt64);
            skUInt64: Result := (A.Value.AsUInt64 > B.Value.AsUInt64);
            skSingle: Result := (A.Value.AsSingle > B.Value.AsSingle);
            skDouble: Result := (A.Value.AsDouble > B.Value.AsDouble);
            skDateTime: Result := (A.Value.AsDateTime > B.Value.AsDateTime);
            skDecimal: Result := (A.Value.AsDecimal > B.Value.AsDecimal);
            skChar: Result := (A.Value.AsChar > B.Value.AsChar);
            skString: Result := (A.Value.AsString > B.Value.AsString);
        else
            raise EArgumentException.CreateFmt('Operator GreaterThan not supported for type %s', [A.Kind.ToString]);
        end;
        exit;
    end;

    // Slow path for mixed, but compatible, types
    const FLOAT_KINDS: set of TScalarKind = [skSingle, skDouble];
    const NUMERIC_KINDS: set of TScalarKind = FLOAT_KINDS + [skInteger, skInt64, skUInt64, skDecimal];

    if not ((A.Kind in NUMERIC_KINDS) and (B.Kind in NUMERIC_KINDS)) then
        raise EArgumentException.CreateFmt('Cannot compare %s and %s', [A.Kind.ToString, B.Kind.ToString]);

    if ((A.Kind = skDecimal) and (B.Kind in FLOAT_KINDS)) or ((B.Kind = skDecimal) and (A.Kind in FLOAT_KINDS)) then
        raise EArgumentException.CreateFmt('Cannot compare Decimal and floating-point types (%s, %s)', [A.Kind.ToString, B.Kind.ToString]);

    if (A.Kind = skDecimal) or (B.Kind = skDecimal) then
    begin
        var valA, valB: TDecimal;
        case A.Kind of
            skInteger: valA := TDecimal.Create(A.Value.AsInteger, 0);
            skInt64: valA := TDecimal.Create(A.Value.AsInt64, 0);
            skUInt64: valA := TDecimal.Create(Int64(A.Value.AsUInt64), 0);
            skDecimal: valA := A.Value.AsDecimal;
        else
            raise EArgumentException.CreateFmt('Internal error comparing %s', [A.Kind.ToString]);
        end;
        case B.Kind of
            skInteger: valB := TDecimal.Create(B.Value.AsInteger, 0);
            skInt64: valB := TDecimal.Create(B.Value.AsInt64, 0);
            skUInt64: valB := TDecimal.Create(Int64(B.Value.AsUInt64), 0);
            skDecimal: valB := B.Value.AsDecimal;
        else
            raise EArgumentException.CreateFmt('Internal error comparing %s', [B.Kind.ToString]);
        end;
        Result := (valA > valB);
    end
    else
    begin
        var dblA, dblB: Double;
        dblA := 0.0;
        dblB := 0.0;
        case A.Kind of
            skInteger: dblA := A.Value.AsInteger;
            skInt64: dblA := A.Value.AsInt64;
            skUInt64: dblA := A.Value.AsUInt64;
            skSingle: dblA := A.Value.AsSingle;
            skDouble: dblA := A.Value.AsDouble;
        end;
        case B.Kind of
            skInteger: dblB := B.Value.AsInteger;
            skInt64: dblB := B.Value.AsInt64;
            skUInt64: dblB := B.Value.AsUInt64;
            skSingle: dblB := B.Value.AsSingle;
            skDouble: dblB := B.Value.AsDouble;
        end;
        Result := (dblA > dblB);
    end;
end;

class operator TScalar.GreaterThanOrEqual(const A, B: TScalar): Boolean;
begin
    Result := (A > B) or (A = B);
end;

class operator TScalar.LessThan(const A, B: TScalar): Boolean;
begin
    // Re-use GreaterThan logic to avoid code duplication
    Result := B > A;
end;

class operator TScalar.LessThanOrEqual(const A, B: TScalar): Boolean;
begin
    Result := (A < B) or (A = B);
end;

class operator TScalar.LogicalNot(const A: TScalar): TScalar;
begin
    Result.Kind := A.Kind;
    case A.Kind of
        skBoolean: Result.Value.AsBoolean := not A.Value.AsBoolean;
        skInteger: Result.Value.AsInteger := not A.Value.AsInteger;
        skInt64: Result.Value.AsInt64 := not A.Value.AsInt64;
        skUInt64: Result.Value.AsUInt64 := not A.Value.AsUInt64;
    else
        raise EArgumentException.CreateFmt('Operator Not not supported for type %s', [A.Kind.ToString]);
    end;
end;

class operator TScalar.Multiply(const A, B: TScalar): TScalar;
begin
    // Fast path for identical types
    if (A.Kind = B.Kind) then
    begin
        Result.Kind := A.Kind;
        case A.Kind of
            skInteger: Result.Value.AsInteger := A.Value.AsInteger * B.Value.AsInteger;
            skInt64: Result.Value.AsInt64 := A.Value.AsInt64 * B.Value.AsInt64;
            skUInt64: Result.Value.AsUInt64 := A.Value.AsUInt64 * B.Value.AsUInt64;
            skSingle: Result.Value.AsSingle := A.Value.AsSingle * B.Value.AsSingle;
            skDouble: Result.Value.AsDouble := A.Value.AsDouble * B.Value.AsDouble;
            skDecimal: Result.Value.AsDecimal := A.Value.AsDecimal * B.Value.AsDecimal;
        else
            raise EArgumentException.CreateFmt('Operator Multiply not supported for type %s', [A.Kind.ToString]);
        end;
        exit;
    end;

    // Slow path for mixed types
    const FLOAT_KINDS: set of TScalarKind = [skSingle, skDouble];
    const NUMERIC_KINDS: set of TScalarKind = FLOAT_KINDS + [skInteger, skInt64, skUInt64, skDecimal];

    if not ((A.Kind in NUMERIC_KINDS) and (B.Kind in NUMERIC_KINDS)) then
        raise EArgumentException.CreateFmt('Operator Multiply not supported for types %s and %s', [A.Kind.ToString, B.Kind.ToString]);

    if (A.Kind = skDecimal) or (B.Kind = skDecimal) then
    begin
        if (A.Kind in FLOAT_KINDS) or (B.Kind in FLOAT_KINDS) then
            raise EArgumentException
                .CreateFmt('Operator Multiply cannot mix Decimal and floating-point types (%s, %s)', [A.Kind.ToString, B.Kind.ToString]);
        var valA, valB: TDecimal;
        case A.Kind of
            skInteger: valA := TDecimal.Create(A.Value.AsInteger, 0);
            skInt64: valA := TDecimal.Create(A.Value.AsInt64, 0);
            skUInt64: valA := TDecimal.Create(Int64(A.Value.AsUInt64), 0);
            skDecimal: valA := A.Value.AsDecimal;
        end;
        case B.Kind of
            skInteger: valB := TDecimal.Create(B.Value.AsInteger, 0);
            skInt64: valB := TDecimal.Create(B.Value.AsInt64, 0);
            skUInt64: valB := TDecimal.Create(Int64(B.Value.AsUInt64), 0);
            skDecimal: valB := B.Value.AsDecimal;
        end;
        Result := TScalar.FromDecimal(valA * valB);
    end
    else if (A.Kind = skDouble) or (B.Kind = skDouble) then
    begin
        var dblA: Double := 0.0;
        var dblB: Double := 0.0;
        case A.Kind of
            skInteger: dblA := A.Value.AsInteger;
            skInt64: dblA := A.Value.AsInt64;
            skUInt64: dblA := A.Value.AsUInt64;
            skSingle: dblA := A.Value.AsSingle;
            skDouble: dblA := A.Value.AsDouble;
        end;
        case B.Kind of
            skInteger: dblB := B.Value.AsInteger;
            skInt64: dblB := B.Value.AsInt64;
            skUInt64: dblB := B.Value.AsUInt64;
            skSingle: dblB := B.Value.AsSingle;
            skDouble: dblB := B.Value.AsDouble;
        end;
        Result := TScalar.FromDouble(dblA * dblB);
    end
    else if (A.Kind = skSingle) or (B.Kind = skSingle) then
    begin
        var sngA: Single := 0.0;
        var sngB: Single := 0.0;
        case A.Kind of
            skInteger: sngA := A.Value.AsInteger;
            skInt64: sngA := A.Value.AsInt64;
            skUInt64: sngA := A.Value.AsUInt64;
            skSingle: sngA := A.Value.AsSingle;
        end;
        case B.Kind of
            skInteger: sngB := B.Value.AsInteger;
            skInt64: sngB := B.Value.AsInt64;
            skUInt64: sngB := B.Value.AsUInt64;
            skSingle: sngB := B.Value.AsSingle;
        end;
        Result := TScalar.FromSingle(sngA * sngB);
    end
    else if (A.Kind = skUInt64) or (B.Kind = skUInt64) then
    begin
        var u64A: UInt64 := 0;
        var u64B: UInt64 := 0;
        case A.Kind of
            skInteger: u64A := UInt64(A.Value.AsInteger);
            skInt64: u64A := UInt64(A.Value.AsInt64);
            skUInt64: u64A := A.Value.AsUInt64;
        end;
        case B.Kind of
            skInteger: u64B := UInt64(B.Value.AsInteger);
            skInt64: u64B := UInt64(B.Value.AsInt64);
            skUInt64: u64B := B.Value.AsUInt64;
        end;
        Result := TScalar.FromUInt64(u64A * u64B);
    end
    else if (A.Kind = skInt64) or (B.Kind = skInt64) then
    begin
        var i64A: Int64 := 0;
        var i64B: Int64 := 0;
        case A.Kind of
            skInteger: i64A := A.Value.AsInteger;
            skInt64: i64A := A.Value.AsInt64;
        end;
        case B.Kind of
            skInteger: i64B := B.Value.AsInteger;
            skInt64: i64B := B.Value.AsInt64;
        end;
        Result := TScalar.FromInt64(i64A * i64B);
    end
    else
    begin
        Result := TScalar.FromInteger(A.Value.AsInteger * B.Value.AsInteger);
    end;
end;

class operator TScalar.Negative(const A: TScalar): TScalar;
begin
    Result.Kind := A.Kind;
    case A.Kind of
        skInteger: Result.Value.AsInteger := -A.Value.AsInteger;
        skInt64: Result.Value.AsInt64 := -A.Value.AsInt64;
        skSingle: Result.Value.AsSingle := -A.Value.AsSingle;
        skDouble: Result.Value.AsDouble := -A.Value.AsDouble;
        skDecimal: Result.Value.AsDecimal := -A.Value.AsDecimal;
    else
        raise EArgumentException.CreateFmt('Operator Negative not supported for type %s', [A.Kind.ToString]);
    end;
end;

class operator TScalar.NotEqual(const A, B: TScalar): Boolean;
begin
    Result := not (A = B);
end;

class operator TScalar.Round(const A: TScalar): TScalar;
begin
    case A.Kind of
        skSingle: Result := TScalar.FromInt64(System.Round(A.Value.AsSingle));
        skDouble: Result := TScalar.FromInt64(System.Round(A.Value.AsDouble));
        skDateTime: Result := TScalar.FromInt64(System.Round(A.Value.AsDateTime));
    else
        raise EArgumentException.CreateFmt('Operator Round not supported for type %s', [A.Kind.ToString]);
    end;
end;

class operator TScalar.Subtract(const A, B: TScalar): TScalar;
begin
    // Fast path for identical types
    if (A.Kind = B.Kind) then
    begin
        if (A.Kind = skDateTime) then
        begin
            Result := TScalar.FromDouble(A.Value.AsDateTime - B.Value.AsDateTime);
            exit;
        end;
        Result.Kind := A.Kind;
        case A.Kind of
            skInteger: Result.Value.AsInteger := A.Value.AsInteger - B.Value.AsInteger;
            skInt64: Result.Value.AsInt64 := A.Value.AsInt64 - B.Value.AsInt64;
            skUInt64: Result.Value.AsUInt64 := A.Value.AsUInt64 - B.Value.AsUInt64;
            skSingle: Result.Value.AsSingle := A.Value.AsSingle - B.Value.AsSingle;
            skDouble: Result.Value.AsDouble := A.Value.AsDouble - B.Value.AsDouble;
            skDecimal: Result.Value.AsDecimal := A.Value.AsDecimal - B.Value.AsDecimal;
        else
            raise EArgumentException.CreateFmt('Operator Subtract not supported for type %s', [A.Kind.ToString]);
        end;
        exit;
    end;

    // Slow path for mixed types
    const FLOAT_KINDS: set of TScalarKind = [skSingle, skDouble];
    const ORDINAL_KINDS: set of TScalarKind = [skInteger, skInt64, skUInt64];
    const NUMERIC_KINDS: set of TScalarKind = FLOAT_KINDS + ORDINAL_KINDS + [skDecimal];

    if not (((A.Kind in NUMERIC_KINDS) and (B.Kind in NUMERIC_KINDS))
        or ((A.Kind = skDateTime) and (B.Kind in NUMERIC_KINDS + [skDateTime]))) then
        raise EArgumentException.CreateFmt('Operator Subtract not supported for types %s and %s', [A.Kind.ToString, B.Kind.ToString]);

    if (A.Kind = skDecimal) or (B.Kind = skDecimal) then
    begin
        if (A.Kind in FLOAT_KINDS) or (B.Kind in FLOAT_KINDS) then
            raise EArgumentException
                .CreateFmt('Operator Subtract cannot mix Decimal and floating-point types (%s, %s)', [A.Kind.ToString, B.Kind.ToString]);
        var valA, valB: TDecimal;
        case A.Kind of
            skInteger: valA := TDecimal.Create(A.Value.AsInteger, 0);
            skInt64: valA := TDecimal.Create(A.Value.AsInt64, 0);
            skUInt64: valA := TDecimal.Create(Int64(A.Value.AsUInt64), 0);
            skDecimal: valA := A.Value.AsDecimal;
        end;
        case B.Kind of
            skInteger: valB := TDecimal.Create(B.Value.AsInteger, 0);
            skInt64: valB := TDecimal.Create(B.Value.AsInt64, 0);
            skUInt64: valB := TDecimal.Create(Int64(B.Value.AsUInt64), 0);
            skDecimal: valB := B.Value.AsDecimal;
        end;
        Result := TScalar.FromDecimal(valA - valB);
    end
    else if (A.Kind = skDateTime) then
    begin
        var numPart: Double := 0.0;
        case B.Kind of
            skInteger: numPart := B.Value.AsInteger;
            skInt64: numPart := B.Value.AsInt64;
            skUInt64: numPart := B.Value.AsUInt64;
            skSingle: numPart := B.Value.AsSingle;
            skDouble: numPart := B.Value.AsDouble;
            skDateTime: numPart := B.Value.AsDateTime;
        end;
        if B.Kind = skDateTime then
            Result := TScalar.FromDouble(A.Value.AsDateTime - numPart)
        else
            Result := TScalar.FromDateTime(A.Value.AsDateTime - numPart);
    end
    else if (B.Kind = skDateTime) then
    begin
        raise EArgumentException.CreateFmt('Operator Subtract not supported for types %s and %s', [A.Kind.ToString, B.Kind.ToString]);
    end
    else if (A.Kind = skDouble) or (B.Kind = skDouble) then
    begin
        var dblA: Double := 0.0;
        var dblB: Double := 0.0;
        case A.Kind of
            skInteger: dblA := A.Value.AsInteger;
            skInt64: dblA := A.Value.AsInt64;
            skUInt64: dblA := A.Value.AsUInt64;
            skSingle: dblA := A.Value.AsSingle;
            skDouble: dblA := A.Value.AsDouble;
        end;
        case B.Kind of
            skInteger: dblB := B.Value.AsInteger;
            skInt64: dblB := B.Value.AsInt64;
            skUInt64: dblB := B.Value.AsUInt64;
            skSingle: dblB := B.Value.AsSingle;
            skDouble: dblB := B.Value.AsDouble;
        end;
        Result := TScalar.FromDouble(dblA - dblB);
    end
    else if (A.Kind = skSingle) or (B.Kind = skSingle) then
    begin
        var sngA: Single := 0.0;
        var sngB: Single := 0.0;
        case A.Kind of
            skInteger: sngA := A.Value.AsInteger;
            skInt64: sngA := A.Value.AsInt64;
            skUInt64: sngA := A.Value.AsUInt64;
            skSingle: sngA := A.Value.AsSingle;
        end;
        case B.Kind of
            skInteger: sngB := B.Value.AsInteger;
            skInt64: sngB := B.Value.AsInt64;
            skUInt64: sngB := B.Value.AsUInt64;
            skSingle: sngB := B.Value.AsSingle;
        end;
        Result := TScalar.FromSingle(sngA - sngB);
    end
    else if (A.Kind = skUInt64) or (B.Kind = skUInt64) then
    begin
        var u64A: UInt64 := 0;
        var u64B: UInt64 := 0;
        case A.Kind of
            skInteger: u64A := UInt64(A.Value.AsInteger);
            skInt64: u64A := UInt64(A.Value.AsInt64);
            skUInt64: u64A := A.Value.AsUInt64;
        end;
        case B.Kind of
            skInteger: u64B := UInt64(B.Value.AsInteger);
            skInt64: u64B := UInt64(B.Value.AsInt64);
            skUInt64: u64B := B.Value.AsUInt64;
        end;
        Result := TScalar.FromUInt64(u64A - u64B);
    end
    else if (A.Kind = skInt64) or (B.Kind = skInt64) then
    begin
        var i64A: Int64 := 0;
        var i64B: Int64 := 0;
        case A.Kind of
            skInteger: i64A := A.Value.AsInteger;
            skInt64: i64A := A.Value.AsInt64;
        end;
        case B.Kind of
            skInteger: i64B := B.Value.AsInteger;
            skInt64: i64B := B.Value.AsInt64;
        end;
        Result := TScalar.FromInt64(i64A - i64B);
    end
    else
    begin
        Result := TScalar.FromInteger(A.Value.AsInteger - B.Value.AsInteger);
    end;
end;

class operator TScalar.Trunc(const A: TScalar): TScalar;
begin
    case A.Kind of
        skSingle: Result := TScalar.FromInt64(System.Trunc(A.Value.AsSingle));
        skDouble: Result := TScalar.FromInt64(System.Trunc(A.Value.AsDouble));
        skDateTime: Result := TScalar.FromInt64(System.Trunc(A.Value.AsDateTime));
    else
        raise EArgumentException.CreateFmt('Operator Trunc not supported for type %s', [A.Kind.ToString]);
    end;
end;

{ TScalarArray }

constructor TScalarArray.Create(AKind: TScalarKind; const AItems: TArray<TScalarValue>);
begin
    Kind := AKind;
    Items := AItems;
end;

{ TScalarRecordField }

constructor TScalarRecordField.Create(const AName: String; AKind: TScalarKind);
begin
    Name := AName;
    Kind := AKind;
end;

{ TScalarRecord }

constructor TScalarRecord.Create(const ADef: TScalarRecordDefinition; const AFields: TArray<TScalarValue>);
begin
    Assert(Length(ADef.Fields) = Length(AFields), 'Field definition and value count must match.');
    FDef := ADef;
    FFields := AFields;
end;

function TScalarRecord.GetDef: TScalarRecordDefinition;
begin
    Result := FDef;
end;

function TScalarRecord.GetFields: TArray<TScalarValue>;
begin
    Result := FFields;
end;

function TScalarRecord.GetItems(const Name: String): TScalar;
var
    idx: Integer;
begin
    idx := FDef.IndexOf(Name);
    if idx < 0 then
        raise EArgumentException.CreateFmt('Field "%s" not found in record definition.', [Name]);

    Result.Create(FDef.Fields[idx].Kind, FFields[idx]);
end;

{ TScalarRecordDefinition }

constructor TScalarRecordDefinition.Create(const AFields: TArray<TScalarRecordField>);
begin
    FFields := AFields;
end;

function TScalarRecordDefinition.IndexOf(const Name: String): Integer;
var
    i: Integer;
begin
    for i := 0 to High(FFields) do
    begin
        if (CompareText(FFields[i].Name, Name) = 0) then
            exit(i);
    end;
    Result := -1;
end;

{ TScalarRecordSeries }

// Implements a time series of scalar records using a chunk array for efficient storage.
// Each record's fields are stored sequentially in the TChunkArray.

constructor TScalarRecordSeries.Create(const ADef: TScalarRecordDefinition);
begin
    Assert(Length(ADef.Fields) > 0);
    FDef := ADef;
    FTotalCount := 0;
end;

procedure TScalarRecordSeries.Add(const Item: TScalarRecord; Lookback: Int64 = -1);
begin
    FArray.Add(Item.Fields, Length(FDef.Fields) * Lookback);
    inc(FTotalCount);
end;

function TScalarRecordSeries.CreateMemberSeries(const Field: String): ISeries;
begin
    var elem := FDef.IndexOf(Field);
    if elem < 0 then
        raise EArgumentException.CreateFmt('Field "%s" not found in record definition.', [Field]);

    Result := TMemberSeries.Create(Self, elem);
end;

function TScalarRecordSeries.GetCount: Int64;
begin
    Result := FArray.Count div Length(FDef.Fields);
end;

function TScalarRecordSeries.GetDef: TScalarRecordDefinition;
begin
    Result := FDef;
end;

function TScalarRecordSeries.GetItemRef(Idx: Integer): TChunkArray<TScalarValue>.PT;
begin
    var len := Length(FDef.Fields);
    Result := FArray.ItemRef[(FArray.Count - len) - (Idx * len)];
end;

function TScalarRecordSeries.GetItems(Idx: Integer): TScalarRecord;
var
    values: TArray<TScalarValue>;
begin
    SetLength(values, Length(FDef.Fields));
    Move(GetItemRef(Idx)^, values[0], sizeof(TScalarValue) * Length(values));
    Result.Create(FDef, values);
end;

function TScalarRecordSeries.GetTotalCount: Int64;
begin
    Result := FTotalCount;
end;

function TScalarKindHelper.ToString: string;
begin
    case Self of
        skInteger: Result := 'integer';
        skInt64: Result := 'int64';
        skUInt64: Result := 'uint64';
        skSingle: Result := 'single';
        skDouble: Result := 'double';
        skDateTime: Result := 'datetime';
        skTimestamp: Result := 'timestamp';
        skBoolean: Result := 'boolean';
        skChar: Result := 'char';
        skPChar: Result := 'pchar';
        skString: Result := 'string';
        skBytes: Result := 'bytes';
        skDecimal: Result := 'decimal';
    else
        Result := 'unknown';
    end;
end;

{ TBinaryOperatorHelper }

function TBinaryOperatorHelper.ToString: string;
begin
    case Self of
        boAdd: Result := '+';
        boSubtract: Result := '-';
        boMultiply: Result := '*';
        boDivide: Result := '/';
        boEqual: Result := '==';
        boNotEqual: Result := '!=';
        boLess: Result := '<';
        boGreater: Result := '>';
        boLessOrEqual: Result := '<=';
        boGreaterOrEqual: Result := '>=';
    else
        Result := '?';
    end;
end;

{ TUnaryOperatorHelper }

function TUnaryOperatorHelper.ToString: string;
begin
    case Self of
        uoNegate: Result := '-';
        uoNot: Result := 'not';
    else
        Result := '?';
    end;
end;

constructor TScalarRecordSeries.TMemberSeries.Create(ARecordSeries: TScalarRecordSeries; AElementIdx: Integer);
begin
    inherited Create;
    FRecordSeries := ARecordSeries;
    FRecordSeries._AddRef;
    FKind := FRecordSeries.FDef.FFields[AElementIdx].Kind;
    FOffset := sizeof(TScalarValue) * AElementIdx;
end;

destructor TScalarRecordSeries.TMemberSeries.Destroy;
begin
    FRecordSeries._Release;
    inherited;
end;

function TScalarRecordSeries.TMemberSeries.GetCount: Int64;
begin
    Result := FRecordSeries.Count;
end;

function TScalarRecordSeries.TMemberSeries.GetItems(Idx: Integer): TScalar;
var
    P: TChunkArray<TScalarValue>.PT;
begin
    P := FRecordSeries.GetItemRef(Idx);
    inc(PByte(P), FOffset);
    Result.Create(FKind, P^);
end;

function TScalarRecordSeries.TMemberSeries.GetTotalCount: Int64;
begin
    Result := FRecordSeries.TotalCount;
end;

{ TScalarSeries }

// Implements a time series of scalar records using a chunk array for efficient storage.
// Each record's fields are stored sequentially in the TChunkArray.

constructor TScalarSeries.Create(AKind: TScalarKind);
begin
    FKind := AKind;
    FTotalCount := 0;
end;

procedure TScalarSeries.Add(const Item: TScalarValue; Lookback: Int64 = -1);
begin
    FArray.Add(Item, Lookback);
    inc(FTotalCount);
end;

function TScalarSeries.GetCount: Int64;
begin
    Result := FArray.Count;
end;

function TScalarSeries.GetItems(Idx: Integer): TScalar;
begin
    Result.Create(FKind, FArray[FArray.Count - Idx - 1]);
end;

function TScalarSeries.GetTotalCount: Int64;
begin
    Result := FTotalCount;
end;

{ TIndexSeries }

constructor TIndexSeries.Create(const AIndexArray: TArray<Integer>);
begin
    inherited Create;
    FIndexArray := AIndexArray;
end;

function TIndexSeries.GetCount: Int64;
begin
    Result := Length(FIndexArray);
end;

function TIndexSeries.GetTotalCount: Int64;
begin
    Result := GetCount;
end;

function TIndexSeries.GetItems(Idx: Integer): TScalar;
var
    sourceIndex: Integer;
begin
    // The series maintains the "0 is newest" convention.
    // The index array was built from oldest to newest, so we access it in reverse.
    sourceIndex := FIndexArray[GetCount - 1 - Idx];
    Result := TScalar.FromInteger(sourceIndex);
end;

initialization
    Assert(sizeof(TScalarValue) = 8);

end.