Fix in DataFeed

This commit is contained in:
Michael Schimmel
2025-12-10 12:19:49 +01:00
parent 235de7a7c5
commit a3f6f4af26
4 changed files with 53 additions and 68 deletions
+1 -1
View File
@@ -4,7 +4,7 @@
<ProjectVersion>20.3</ProjectVersion>
<FrameworkType>FMX</FrameworkType>
<Base>True</Base>
<Config Condition="'$(Config)'==''">Release</Config>
<Config Condition="'$(Config)'==''">Debug</Config>
<Platform Condition="'$(Platform)'==''">Win64</Platform>
<ProjectName Condition="'$(ProjectName)'==''">AuraTrader</ProjectName>
<TargetedPlatforms>3</TargetedPlatforms>
+1 -1
View File
@@ -775,7 +775,7 @@ begin
if allScalar then
begin
def := TScalarRecordRegistry.Intern(scalarDefFields);
def := TScalarRecord.TRegistry.Intern(scalarDefFields);
staticType := TTypes.CreateRecord(def);
Result := TAst.RecordLiteral(Node.Identity, fieldList, def, nil, staticType);
end
+1 -1
View File
@@ -63,7 +63,7 @@ begin
fields[i] := TScalarRecordField.Create(fieldKey, TScalar.TKind(GetEnumValue(TypeInfo(TScalar.TKind), kindStr)));
end;
Result := TScalarRecordRegistry.Intern(fields);
Result := TScalarRecord.TRegistry.Intern(fields);
finally
jsonValue.Free;
end;
+50 -65
View File
@@ -53,8 +53,8 @@ type
function GetIsValid: Boolean;
public
constructor Create(const APath, ASymbol, AExtension: String; AYear, AMonth: Integer; const ABasename: String);
function GetBaseFileName: string;
function GetFullFileName: string;
property BaseFileName: string read FBasename;
property IsValid: Boolean read GetIsValid;
property Extension: String read FExtension;
property Path: String read FPath;
@@ -84,10 +84,14 @@ type
// Loading Logic
function DoLoad(const FileName: string; const Gate: TLatch): TFuture<TArray<TBatch>>;
function LoadDataFile(const DataFile: TDataFile): TFuture<TArray<TBatch>>;
function GetNextFile(const Curr: TDataFile): TDataFile;
// Processing Logic
function ProcessFile(FileInfo: TDataFile; const DataFile: TFuture<TArray<TBatch>>; const Terminated: TState): TState;
function ProcessFile(
const Files: TArray<TDataFile>;
Index: Integer;
const DataFile: TFuture<TArray<TBatch>>;
const Terminated: TState
): TState;
function ProcessChunks(const Batches: TArray<TBatch>; Terminated: TState): TState;
@@ -97,7 +101,6 @@ type
// Helpers
function ParseFileName(const FileName: string): TDataFile;
function FindFirstFile(const Symbol: string): TFuture<TDataFile>;
procedure UpdateSymbols;
private
@@ -213,14 +216,9 @@ begin
FBasename := ABasename;
end;
function TScalarRecordFeed.TDataFile.GetBaseFileName: string;
begin
Result := FBasename;
end;
function TScalarRecordFeed.TDataFile.GetFullFileName: string;
begin
Result := TPath.Combine(FPath, GetBaseFileName + FExtension);
Result := TPath.Combine(FPath, FBasename + FExtension);
end;
function TScalarRecordFeed.TDataFile.GetIsValid: Boolean;
@@ -350,51 +348,6 @@ begin
);
end;
function TScalarRecordFeed.FindFirstFile(const Symbol: string): TFuture<TDataFile>;
begin
var symName := Symbol;
Result :=
FSymbols.Chain<TDataFile>(
function(const Symbols: TDictionary<String, TArray<TDataFile>>): TDataFile
var
symbolFiles: TArray<TDataFile>;
begin
if Symbols.TryGetValue(symName, symbolFiles) and (Length(symbolFiles) > 0) then
Result := symbolFiles[0]
else
Result := Default(TDataFile);
end
);
end;
function TScalarRecordFeed.GetNextFile(const Curr: TDataFile): TDataFile;
var
allSymbolFiles: TArray<TDataFile>;
foundIndex: Integer;
begin
Result := Default(TDataFile);
if not Curr.IsValid or not FSymbols.Done.IsSet then
exit;
if not FSymbols.Value.TryGetValue(Curr.Symbol, allSymbolFiles) then
exit;
foundIndex := -1;
for var i := 0 to High(allSymbolFiles) do
begin
if allSymbolFiles[i].GetBaseFileName() = Curr.GetBaseFileName() then
begin
foundIndex := i;
break;
end;
end;
if (foundIndex <> -1) and (foundIndex < High(allSymbolFiles)) then
begin
Result := allSymbolFiles[foundIndex + 1];
end;
end;
function TScalarRecordFeed.ParseFileName(const FileName: string): TDataFile;
var
path, fileNameNoPath, baseName, ext, symbol: string;
@@ -608,32 +561,64 @@ end;
function TScalarRecordFeed.ProcessData(const Symbol: String; const Terminated: TState): TState;
begin
var cTerminated := Terminated;
var cSymbol := Symbol;
// We fetch the list once, then iterate by index.
Result :=
FindFirstFile(Symbol)
.Chain(
function(const FirstInfo: TDataFile): TState
FSymbols.Chain(
function(const AllSymbols: TDictionary<String, TArray<TDataFile>>): TState
var
files: TArray<TDataFile>;
begin
if AllSymbols.TryGetValue(cSymbol, files) and (Length(files) > 0) then
begin
Result := ProcessFile(FirstInfo, LoadDataFile(FirstInfo), cTerminated);
end);
// Start with Index 0
Result := ProcessFile(files, 0, LoadDataFile(files[0]), cTerminated);
end
else
begin
Result := TState.Null;
end;
end
);
end;
function TScalarRecordFeed.ProcessFile(FileInfo: TDataFile; const DataFile: TFuture<TArray<TBatch>>; const Terminated: TState): TState;
function TScalarRecordFeed.ProcessFile(
const Files: TArray<TDataFile>;
Index: Integer;
const DataFile: TFuture<TArray<TBatch>>;
const Terminated: TState
): TState;
begin
if not FileInfo.IsValid or Terminated.IsSet then
// Check bounds and termination
if (Index < 0) or (Index > High(Files)) or Terminated.IsSet then
exit(DataFile.Done);
var nextInfo := GetNextFile(FileInfo);
var nextFuture := LoadDataFile(nextInfo);
var currentInfo := Files[Index];
if not currentInfo.IsValid then
exit(DataFile.Done);
// Prefetch logic: Prepare the next file immediately (if available)
var nextIndex := Index + 1;
var nextFuture: TFuture<TArray<TBatch>>;
if nextIndex <= High(Files) then
nextFuture := LoadDataFile(Files[nextIndex])
else
nextFuture := TFuture<TArray<TBatch>>.Construct(TArray<TBatch>(nil)); // End of stream
var cTerminated := Terminated;
// Process current data, then recurse to next index
Result :=
DataFile.Chain(
function(const Batches: TArray<TBatch>): TState
begin
var done := ProcessChunks(Batches, Terminated);
Result := TaskManager.RunTask(done, function: TState begin Result := ProcessFile(nextInfo, nextFuture, cTerminated); end);
// Recurse with Index + 1
Result :=
TaskManager.RunTask(done, function: TState begin Result := ProcessFile(Files, nextIndex, nextFuture, cTerminated); end);
end
);
end;