diff --git a/csharp/ExcelAddIn/DeephavenExcelFunctions.cs b/csharp/ExcelAddIn/DeephavenExcelFunctions.cs index d80d6627c85..9aeab0ad654 100644 --- a/csharp/ExcelAddIn/DeephavenExcelFunctions.cs +++ b/csharp/ExcelAddIn/DeephavenExcelFunctions.cs @@ -1,12 +1,8 @@ -using System.Diagnostics; -using System.Diagnostics.CodeAnalysis; +using System.Diagnostics.CodeAnalysis; using Deephaven.ExcelAddIn.ExcelDna; using Deephaven.ExcelAddIn.Factories; using Deephaven.ExcelAddIn.Models; using Deephaven.ExcelAddIn.Operations; -using Deephaven.ExcelAddIn.Providers; -using Deephaven.ExcelAddIn.Viewmodels; -using Deephaven.ExcelAddIn.Views; using ExcelDna.Integration; namespace Deephaven.ExcelAddIn; @@ -21,38 +17,38 @@ public static void ShowConnectionsDialog() { [ExcelFunction(Description = "Snapshots a table", IsThreadSafe = true)] public static object DEEPHAVEN_SNAPSHOT(string tableDescriptor, object filter, object wantHeaders) { - if (!TryInterpretCommonArgs(tableDescriptor, filter, wantHeaders, out var td, out var filt, out var wh, out var errorText)) { + if (!TryInterpretCommonArgs(tableDescriptor, filter, wantHeaders, out var tq, out var wh, out var errorText)) { return errorText; } // These two are used by ExcelDNA to share results for identical invocations. The functionName is arbitary but unique. const string functionName = "Deephaven.ExcelAddIn.DeephavenExcelFunctions.DEEPHAVEN_SNAPSHOT"; var parms = new[] { tableDescriptor, filter, wantHeaders }; - ExcelObservableSource eos = () => new SnapshotOperation(td!, filt, wh, StateManager); + ExcelObservableSource eos = () => new SnapshotOperation(tq, wh, StateManager); return ExcelAsyncUtil.Observe(functionName, parms, eos); } [ExcelFunction(Description = "Subscribes to a table", IsThreadSafe = true)] public static object DEEPHAVEN_SUBSCRIBE(string tableDescriptor, object filter, object wantHeaders) { - if (!TryInterpretCommonArgs(tableDescriptor, filter, wantHeaders, out var td, out var filt, out var wh, out string errorText)) { + if (!TryInterpretCommonArgs(tableDescriptor, filter, wantHeaders, out var tq, out var wh, out string errorText)) { return errorText; } // These two are used by ExcelDNA to share results for identical invocations. The functionName is arbitary but unique. const string functionName = "Deephaven.ExcelAddIn.DeephavenExcelFunctions.DEEPHAVEN_SUBSCRIBE"; var parms = new[] { tableDescriptor, filter, wantHeaders }; - ExcelObservableSource eos = () => new SubscribeOperation(td, filt, wh, StateManager); + ExcelObservableSource eos = () => new SubscribeOperation(tq, wh, StateManager); return ExcelAsyncUtil.Observe(functionName, parms, eos); } private static bool TryInterpretCommonArgs(string tableDescriptor, object filter, object wantHeaders, - [NotNullWhen(true)]out TableTriple? tableDescriptorResult, out string filterResult, out bool wantHeadersResult, out string errorText) { - filterResult = ""; + [NotNullWhen(true)]out TableQuad? tableQuadResult, out bool wantHeadersResult, out string errorText) { + tableQuadResult = null; wantHeadersResult = false; - if (!TableTriple.TryParse(tableDescriptor, out tableDescriptorResult, out errorText)) { + if (!TableTriple.TryParse(tableDescriptor, out var tt, out errorText)) { return false; } - if (!ExcelDnaHelpers.TryInterpretAs(filter, "", out filterResult)) { + if (!ExcelDnaHelpers.TryInterpretAs(filter, "", out var condition)) { errorText = "Can't interpret FILTER argument"; return false; } @@ -62,6 +58,8 @@ private static bool TryInterpretCommonArgs(string tableDescriptor, object filter errorText = "Can't interpret WANT_HEADERS argument"; return false; } + + tableQuadResult = new TableQuad(tt.EndpointId, tt.PersistentQueryId, tt.TableName, condition); return true; } } diff --git a/csharp/ExcelAddIn/ExcelAddIn.csproj.user b/csharp/ExcelAddIn/ExcelAddIn.csproj.user index f39927bb679..476df3e2650 100644 --- a/csharp/ExcelAddIn/ExcelAddIn.csproj.user +++ b/csharp/ExcelAddIn/ExcelAddIn.csproj.user @@ -8,5 +8,8 @@ Form + + Form + \ No newline at end of file diff --git a/csharp/ExcelAddIn/StateManager.cs b/csharp/ExcelAddIn/StateManager.cs index a4bb4e44e4e..9faa64a24ae 100644 --- a/csharp/ExcelAddIn/StateManager.cs +++ b/csharp/ExcelAddIn/StateManager.cs @@ -1,4 +1,5 @@ -using Deephaven.DeephavenClient.ExcelAddIn.Util; +using System.Diagnostics; +using System.Net; using Deephaven.DeephavenClient; using Deephaven.ExcelAddIn.Models; using Deephaven.ExcelAddIn.Providers; @@ -8,76 +9,170 @@ namespace Deephaven.ExcelAddIn; public class StateManager { public readonly WorkerThread WorkerThread = WorkerThread.Create(); - private readonly SessionProviders _sessionProviders; + private readonly Dictionary _credentialsProviders = new(); + private readonly Dictionary _sessionProviders = new(); + private readonly Dictionary _persistentQueryProviders = new(); + private readonly Dictionary _tableProviders = new(); + private readonly ObserverContainer> _credentialsPopulationObservers = new(); + private readonly ObserverContainer _defaultEndpointSelectionObservers = new(); - public StateManager() { - _sessionProviders = new SessionProviders(WorkerThread); + private EndpointId? _defaultEndpointId = null; + + public IDisposable SubscribeToCredentialsPopulation(IObserver> observer) { + WorkerThread.EnqueueOrRun(() => { + _credentialsPopulationObservers.Add(observer, out _); + + // Give this observer the current set of endpoint ids. + var keys = _credentialsProviders.Keys.ToArray(); + foreach (var endpointId in keys) { + observer.OnNext(AddOrRemove.OfAdd(endpointId)); + } + }); + + return WorkerThread.EnqueueOrRunWhenDisposed( + () => _credentialsPopulationObservers.Remove(observer, out _)); + } + + public IDisposable SubscribeToDefaultEndpointSelection(IObserver observer) { + WorkerThread.EnqueueOrRun(() => { + _defaultEndpointSelectionObservers.Add(observer, out _); + observer.OnNext(_defaultEndpointId); + }); + + return WorkerThread.EnqueueOrRunWhenDisposed( + () => _defaultEndpointSelectionObservers.Remove(observer, out _)); } - public IDisposable SubscribeToSessions(IObserver> observer) { - return _sessionProviders.Subscribe(observer); + /// + /// The major difference between the credentials providers and the other providers + /// is that the credential providers don't remove themselves from the map + /// upon the last dispose of the subscriber. That is, they hang around until we + /// manually remove them. + /// + public IDisposable SubscribeToCredentials(EndpointId endpointId, + IObserver> observer) { + IDisposable? disposer = null; + LookupOrCreateCredentialsProvider(endpointId, + cp => disposer = cp.Subscribe(observer)); + + return WorkerThread.EnqueueOrRunWhenDisposed(() => + Utility.Exchange(ref disposer, null)?.Dispose()); } - public IDisposable SubscribeToSession(EndpointId endpointId, IObserver> observer) { - return _sessionProviders.SubscribeToSession(endpointId, observer); + public void SetCredentials(CredentialsBase credentials) { + LookupOrCreateCredentialsProvider(credentials.Id, + cp => cp.SetCredentials(credentials)); } - public IDisposable SubscribeToCredentials(EndpointId endpointId, IObserver> observer) { - return _sessionProviders.SubscribeToCredentials(endpointId, observer); + public void Reconnect(EndpointId id) { + // Quick-and-dirty trick for reconnect is to re-send the credentials to the observers. + LookupOrCreateCredentialsProvider(id, cp => cp.Resend()); } - public IDisposable SubscribeToDefaultSession(IObserver> observer) { - return _sessionProviders.SubscribeToDefaultSession(observer); + public void TryDeleteCredentials(EndpointId id, Action onSuccess, Action onFailure) { + if (WorkerThread.EnqueueOrNop(() => TryDeleteCredentials(id, onSuccess, onFailure))) { + return; + } + + if (!_credentialsProviders.TryGetValue(id, out var cp)) { + onFailure($"{id} unknown"); + return; + } + + if (cp.ObserverCountUnsafe != 0) { + onFailure($"{id} is still active"); + return; + } + + if (id.Equals(_defaultEndpointId)) { + SetDefaultEndpointId(null); + } + + _credentialsProviders.Remove(id); + _credentialsPopulationObservers.OnNext(AddOrRemove.OfRemove(id)); + onSuccess(); } - public IDisposable SubscribeToDefaultCredentials(IObserver> observer) { - return _sessionProviders.SubscribeToDefaultCredentials(observer); + private void LookupOrCreateCredentialsProvider(EndpointId endpointId, + Action action) { + if (WorkerThread.EnqueueOrNop(() => LookupOrCreateCredentialsProvider(endpointId, action))) { + return; + } + if (!_credentialsProviders.TryGetValue(endpointId, out var cp)) { + cp = new CredentialsProvider(this); + _credentialsProviders.Add(endpointId, cp); + cp.Init(); + _credentialsPopulationObservers.OnNext(AddOrRemove.OfAdd(endpointId)); + } + + action(cp); } - public IDisposable SubscribeToTableTriple(TableTriple descriptor, string filter, - IObserver> observer) { - // There is a chain with multiple elements: - // - // 1. Make a TableHandleProvider - // 2. Make a ClientProvider - // 3. Subscribe the ClientProvider to either the session provider named by the endpoint id - // or to the default session provider - // 4. Subscribe the TableHandleProvider to the ClientProvider - // 4. Subscribe our observer to the TableHandleProvider - // 5. Return a dispose action that disposes all the needfuls. - - var thp = new TableHandleProvider(WorkerThread, descriptor, filter); - var cp = new ClientProvider(WorkerThread, descriptor); - - var disposer1 = descriptor.EndpointId == null ? - SubscribeToDefaultSession(cp) : - SubscribeToSession(descriptor.EndpointId, cp); - var disposer2 = cp.Subscribe(thp); - var disposer3 = thp.Subscribe(observer); - - // The disposer for this needs to dispose both "inner" disposers. - return ActionAsDisposable.Create(() => { - // TODO(kosak): probably don't need to be on the worker thread here - WorkerThread.Invoke(() => { - var temp1 = Utility.Exchange(ref disposer1, null); - var temp2 = Utility.Exchange(ref disposer2, null); - var temp3 = Utility.Exchange(ref disposer3, null); - temp3?.Dispose(); - temp2?.Dispose(); - temp1?.Dispose(); - }); + public IDisposable SubscribeToSession(EndpointId endpointId, + IObserver> observer) { + IDisposable? disposer = null; + WorkerThread.EnqueueOrRun(() => { + if (!_sessionProviders.TryGetValue(endpointId, out var sp)) { + sp = new SessionProvider(this, endpointId, () => _sessionProviders.Remove(endpointId)); + _sessionProviders.Add(endpointId, sp); + sp.Init(); + } + disposer = sp.Subscribe(observer); }); + + return WorkerThread.EnqueueOrRunWhenDisposed(() => + Utility.Exchange(ref disposer, null)?.Dispose()); } - public void SetCredentials(CredentialsBase credentials) { - _sessionProviders.SetCredentials(credentials); + public IDisposable SubscribeToPersistentQuery(EndpointId endpointId, PersistentQueryId? pqId, + IObserver> observer) { + + IDisposable? disposer = null; + WorkerThread.EnqueueOrRun(() => { + var key = new PersistentQueryKey(endpointId, pqId); + if (!_persistentQueryProviders.TryGetValue(key, out var pqp)) { + pqp = new PersistentQueryProvider(this, endpointId, pqId, + () => _persistentQueryProviders.Remove(key)); + _persistentQueryProviders.Add(key, pqp); + pqp.Init(); + } + disposer = pqp.Subscribe(observer); + }); + + return WorkerThread.EnqueueOrRunWhenDisposed( + () => Utility.Exchange(ref disposer, null)?.Dispose()); } - public void SetDefaultCredentials(CredentialsBase credentials) { - _sessionProviders.SetDefaultCredentials(credentials); + public IDisposable SubscribeToTable(TableQuad key, IObserver> observer) { + IDisposable? disposer = null; + WorkerThread.EnqueueOrRun(() => { + if (!_tableProviders.TryGetValue(key, out var tp)) { + Action onDispose = () => _tableProviders.Remove(key); + if (key.EndpointId == null) { + tp = new DefaultEndpointTableProvider(this, key.PersistentQueryId, key.TableName, key.Condition, + onDispose); + } else if (key.Condition.Length != 0) { + tp = new FilteredTableProvider(this, key.EndpointId, key.PersistentQueryId, key.TableName, + key.Condition, onDispose); + } else { + tp = new TableProvider(this, key.EndpointId, key.PersistentQueryId, key.TableName, onDispose); + } + _tableProviders.Add(key, tp); + tp.Init(); + } + disposer = tp.Subscribe(observer); + }); + + return WorkerThread.EnqueueOrRunWhenDisposed( + () => Utility.Exchange(ref disposer, null)?.Dispose()); } + + public void SetDefaultEndpointId(EndpointId? defaultEndpointId) { + if (WorkerThread.EnqueueOrNop(() => SetDefaultEndpointId(defaultEndpointId))) { + return; + } - public void Reconnect(EndpointId id) { - _sessionProviders.Reconnect(id); + _defaultEndpointId = defaultEndpointId; + _defaultEndpointSelectionObservers.OnNext(_defaultEndpointId); } } diff --git a/csharp/ExcelAddIn/factories/ConnectionManagerDialogFactory.cs b/csharp/ExcelAddIn/factories/ConnectionManagerDialogFactory.cs index cdf6053f327..72441c93323 100644 --- a/csharp/ExcelAddIn/factories/ConnectionManagerDialogFactory.cs +++ b/csharp/ExcelAddIn/factories/ConnectionManagerDialogFactory.cs @@ -1,102 +1,17 @@ -using Deephaven.ExcelAddIn.Viewmodels; -using Deephaven.ExcelAddIn.ViewModels; -using Deephaven.ExcelAddIn.Views; -using System.Diagnostics; -using Deephaven.ExcelAddIn.Models; +using Deephaven.ExcelAddIn.Managers; using Deephaven.ExcelAddIn.Util; +using Deephaven.ExcelAddIn.Views; namespace Deephaven.ExcelAddIn.Factories; internal static class ConnectionManagerDialogFactory { - public static void CreateAndShow(StateManager sm) { - // The "new" button creates a "New/Edit Credentials" dialog - void OnNewButtonClicked() { - var cvm = CredentialsDialogViewModel.OfEmpty(); - var dialog = CredentialsDialogFactory.Create(sm, cvm); - dialog.Show(); - } - - var cmDialog = new ConnectionManagerDialog(OnNewButtonClicked); - cmDialog.Show(); - var cmso = new ConnectionManagerSessionObserver(sm, cmDialog); - var disposer = sm.SubscribeToSessions(cmso); - - cmDialog.Closed += (_, _) => { - disposer.Dispose(); - cmso.Dispose(); - }; - } -} - -internal class ConnectionManagerSessionObserver( - StateManager stateManager, - ConnectionManagerDialog cmDialog) : IObserver>, IDisposable { - private readonly List _disposables = new(); - - public void OnNext(AddOrRemove aor) { - if (!aor.IsAdd) { - // TODO(kosak) - Debug.WriteLine("Remove is not handled"); - return; - } - - var endpointId = aor.Value; - - var statusRow = new ConnectionManagerDialogRow(endpointId.Id, stateManager); - // We watch for session and credential state changes in our ID - var sessDisposable = stateManager.SubscribeToSession(endpointId, statusRow); - var credDisposable = stateManager.SubscribeToCredentials(endpointId, statusRow); - - // And we also watch for credentials changes in the default session (just to keep - // track of whether we are still the default) - var dct = new DefaultCredentialsTracker(statusRow); - var defaultCredDisposable = stateManager.SubscribeToDefaultCredentials(dct); - - // We'll do our AddRow on the GUI thread, and, while we're on the GUI thread, we'll add - // our disposables to our saved disposables. - cmDialog.Invoke(() => { - _disposables.Add(sessDisposable); - _disposables.Add(credDisposable); - _disposables.Add(defaultCredDisposable); - cmDialog.AddRow(statusRow); + public static void CreateAndShow(StateManager stateManager) { + Utility.RunInBackground(() => { + var cmDialog = new ConnectionManagerDialog(); + var dm = ConnectionManagerDialogManager.Create(stateManager, cmDialog); + cmDialog.Closed += (_, _) => dm.Dispose(); + // Blocks forever (in this private thread) + cmDialog.ShowDialog(); }); } - - public void Dispose() { - // Since the GUI thread is where we added these disposables, the GUI thread is where we will - // access and dispose them. - cmDialog.Invoke(() => { - var temp = _disposables.ToArray(); - _disposables.Clear(); - foreach (var disposable in temp) { - disposable.Dispose(); - } - }); - } - - public void OnCompleted() { - // TODO(kosak) - throw new NotImplementedException(); - } - - public void OnError(Exception error) { - // TODO(kosak) - throw new NotImplementedException(); - } } - -internal class DefaultCredentialsTracker(ConnectionManagerDialogRow statusRow) : IObserver> { - public void OnNext(StatusOr value) { - statusRow.SetDefaultCredentials(value); - } - - public void OnCompleted() { - // TODO(kosak) - throw new NotImplementedException(); - } - - public void OnError(Exception error) { - // TODO(kosak) - throw new NotImplementedException(); - } -} \ No newline at end of file diff --git a/csharp/ExcelAddIn/factories/CredentialsDialogFactory.cs b/csharp/ExcelAddIn/factories/CredentialsDialogFactory.cs index c748847e852..4c1e2694be7 100644 --- a/csharp/ExcelAddIn/factories/CredentialsDialogFactory.cs +++ b/csharp/ExcelAddIn/factories/CredentialsDialogFactory.cs @@ -2,81 +2,137 @@ using Deephaven.ExcelAddIn.Util; using Deephaven.ExcelAddIn.ViewModels; using ExcelAddIn.views; +using static System.Windows.Forms.AxHost; namespace Deephaven.ExcelAddIn.Factories; internal static class CredentialsDialogFactory { - public static CredentialsDialog Create(StateManager sm, CredentialsDialogViewModel cvm) { - CredentialsDialog? credentialsDialog = null; + public static void CreateAndShow(StateManager stateManager, CredentialsDialogViewModel cvm, + EndpointId? whitelistId) { + Utility.RunInBackground(() => { + var cd = new CredentialsDialog(cvm); + var state = new CredentialsDialogState(stateManager, cd, cvm, whitelistId); - void OnSetCredentialsButtonClicked() { - if (!cvm.TryMakeCredentials(out var newCreds, out var error)) { - ShowMessageBox(error); - return; - } - sm.SetCredentials(newCreds); - if (cvm.IsDefault) { - sm.SetDefaultCredentials(newCreds); + cd.OnSetCredentialsButtonClicked += state.OnSetCredentials; + cd.OnTestCredentialsButtonClicked += state.OnTestCredentials; + + cd.Closed += (_, _) => state.Dispose(); + // Blocks forever (in this private thread) + cd.ShowDialog(); + }); + } +} + +internal class CredentialsDialogState : IObserver>, IDisposable { + private readonly StateManager _stateManager; + private readonly CredentialsDialog _credentialsDialog; + private readonly CredentialsDialogViewModel _cvm; + private readonly EndpointId? _whitelistId; + private IDisposable? _disposer; + private readonly object _sync = new(); + private readonly HashSet _knownIds = new(); + private readonly VersionTracker _versionTracker = new(); + + public CredentialsDialogState( + StateManager stateManager, + CredentialsDialog credentialsDialog, + CredentialsDialogViewModel cvm, + EndpointId? whitelistId) { + _stateManager = stateManager; + _credentialsDialog = credentialsDialog; + _cvm = cvm; + _whitelistId = whitelistId; + _disposer = stateManager.SubscribeToCredentialsPopulation(this); + } + + public void Dispose() { + Utility.Exchange(ref _disposer, null)?.Dispose(); + } + + public void OnCompleted() { + throw new NotImplementedException(); + } + + public void OnError(Exception error) { + throw new NotImplementedException(); + } + + public void OnNext(AddOrRemove value) { + lock (_sync) { + if (value.IsAdd) { + _knownIds.Add(value.Value); + } else { + _knownIds.Remove(value.Value); } + } + } - credentialsDialog!.Close(); + public void OnSetCredentials() { + if (!_cvm.TryMakeCredentials(out var newCreds, out var error)) { + ShowMessageBox(error); + return; } - // This is used to ignore the results from stale "Test Credentials" invocations - // and to only use the results from the latest. It is read and written from different - // threads so we protect it with a synchronization object. - var sharedTestCredentialsCookie = new SimpleAtomicReference(new object()); - - void TestCredentials(CredentialsBase creds) { - // Make a unique sentinel object to indicate that this thread should be - // the one privileged to provide the system with the answer to the "Test - // Credentials" question. If the user doesn't press the button again, - // we will go ahead and provide our answer to the system. However, if the - // user presses the button again, triggering a new thread, then that - // new thread will usurp our privilege and it will be the one to provide - // the answer. - var localLatestTcc = new object(); - sharedTestCredentialsCookie.Value = localLatestTcc; - - var state = "OK"; - try { - // This operation might take some time. - var temp = SessionBaseFactory.Create(creds, sm.WorkerThread); - temp.Dispose(); - } catch (Exception ex) { - state = ex.Message; - } + bool isKnown; + lock (_sync) { + isKnown = _knownIds.Contains(newCreds.Id); + } - // If sharedTestCredentialsCookie is still the same, then our privilege - // has not been usurped and we can provide our answer to the system. - // On the other hand, if it changes, then we will just throw away our work. - if (!ReferenceEquals(localLatestTcc, sharedTestCredentialsCookie.Value)) { - // Our results are moot. Dispose of them. + if (isKnown && !newCreds.Id.Equals(_whitelistId)) { + const string caption = "Modify existing connection?"; + var text = $"Are you sure you want to modify connection \"{newCreds.Id}\""; + var dhm = new DeephavenMessageBox(caption, text, true); + var dialogResult = dhm.ShowDialog(_credentialsDialog); + if (dialogResult != DialogResult.OK) { return; } + } - // Our results are valid. Keep them and tell everyone about it. - credentialsDialog!.SetTestResultsBox(state); + _stateManager.SetCredentials(newCreds); + if (_cvm.IsDefault) { + _stateManager.SetDefaultEndpointId(newCreds.Id); } - void OnTestCredentialsButtonClicked() { - if (!cvm.TryMakeCredentials(out var newCreds, out var error)) { - ShowMessageBox(error); - return; - } + _credentialsDialog!.Close(); + } + + public void OnTestCredentials() { + if (!_cvm.TryMakeCredentials(out var newCreds, out var error)) { + ShowMessageBox(error); + return; + } + + _credentialsDialog!.SetTestResultsBox("Checking credentials"); + // Check credentials on its own thread + Utility.RunInBackground(() => TestCredentialsThreadFunc(newCreds)); + } + + private void TestCredentialsThreadFunc(CredentialsBase creds) { + var latestCookie = _versionTracker.SetNewVersion(); + + var state = "OK"; + try { + // This operation might take some time. + var temp = SessionBaseFactory.Create(creds, _stateManager.WorkerThread); + temp.Dispose(); + } catch (Exception ex) { + state = ex.Message; + } - credentialsDialog!.SetTestResultsBox("Checking credentials"); - // Check credentials on its own thread - Utility.RunInBackground(() => TestCredentials(newCreds)); + if (!latestCookie.IsCurrent) { + // Our results are moot. Dispose of them. + return; } - // Save in captured variable so that the lambdas can access it. - credentialsDialog = new CredentialsDialog(cvm, OnSetCredentialsButtonClicked, OnTestCredentialsButtonClicked); - return credentialsDialog; + // Our results are valid. Keep them and tell everyone about it. + _credentialsDialog!.SetTestResultsBox(state); } - private static void ShowMessageBox(string error) { - MessageBox.Show(error, "Please provide missing fields", MessageBoxButtons.OK); + private void ShowMessageBox(string error) { + _credentialsDialog.Invoke(() => { + var dhm = new DeephavenMessageBox("Please provide missing fields", error, false); + dhm.ShowDialog(_credentialsDialog); + }); } } diff --git a/csharp/ExcelAddIn/managers/ConnectionManagerDialogManager.cs b/csharp/ExcelAddIn/managers/ConnectionManagerDialogManager.cs new file mode 100644 index 00000000000..5ebcbc958f6 --- /dev/null +++ b/csharp/ExcelAddIn/managers/ConnectionManagerDialogManager.cs @@ -0,0 +1,197 @@ +using System.Collections.Concurrent; +using Deephaven.ExcelAddIn.Models; +using Deephaven.ExcelAddIn.Viewmodels; +using Deephaven.ExcelAddIn.Views; +using Deephaven.ExcelAddIn.Util; +using Deephaven.ExcelAddIn.Factories; +using Deephaven.ExcelAddIn.ViewModels; +using ExcelAddIn.views; + +namespace Deephaven.ExcelAddIn.Managers; + +internal class ConnectionManagerDialogManager : IObserver>, IDisposable { + // + // ConnectionManagerDialog cmDialog, + // ConcurrentDictionary rowToManager, + // StateManager stateManager) + public static ConnectionManagerDialogManager Create(StateManager stateManager, + ConnectionManagerDialog cmDialog) { + var result = new ConnectionManagerDialogManager(stateManager, cmDialog); + cmDialog.OnNewButtonClicked += result.OnNewButtonClicked; + cmDialog.OnDeleteButtonClicked += result.OnDeleteButtonClicked; + cmDialog.OnReconnectButtonClicked += result.OnReconnectButtonClicked; + cmDialog.OnMakeDefaultButtonClicked += result.OnMakeDefaultButtonClicked; + cmDialog.OnEditButtonClicked += result.OnEditButtonClicked; + + var disp = stateManager.SubscribeToCredentialsPopulation(result); + result._disposables.Add(disp); + return result; + } + + private readonly StateManager _stateManager; + private readonly WorkerThread _workerThread; + private readonly ConnectionManagerDialog _cmDialog; + private readonly Dictionary _idToRow = new(); + private readonly Dictionary _rowToManager = new(); + private readonly List _disposables = new(); + + public ConnectionManagerDialogManager(StateManager stateManager, ConnectionManagerDialog cmDialog) { + _stateManager = stateManager; + _workerThread = stateManager.WorkerThread; + _cmDialog = cmDialog; + } + + public void OnNext(AddOrRemove aor) { + if (_workerThread.EnqueueOrNop(() => OnNext(aor))) { + return; + } + + if (aor.IsAdd) { + var endpointId = aor.Value; + var row = new ConnectionManagerDialogRow(endpointId.Id); + var statusRowManager = ConnectionManagerDialogRowManager.Create(row, endpointId, _stateManager); + _rowToManager.Add(row, statusRowManager); + _idToRow.Add(endpointId, row); + _disposables.Add(statusRowManager); + + _cmDialog.AddRow(row); + return; + } + + // Remove! + if (!_idToRow.Remove(aor.Value, out var rowToDelete) || + !_rowToManager.Remove(rowToDelete, out var rowManager)) { + return; + } + + _cmDialog.RemoveRow(rowToDelete); + rowManager.Dispose(); + } + + public void Dispose() { + if (_workerThread.EnqueueOrNop(Dispose)) { + return; + } + + var temp = _disposables.ToArray(); + _disposables.Clear(); + foreach (var disposable in temp) { + disposable.Dispose(); + } + } + + public void OnCompleted() { + // TODO(kosak) + throw new NotImplementedException(); + } + + public void OnError(Exception error) { + // TODO(kosak) + throw new NotImplementedException(); + } + + void OnNewButtonClicked() { + var cvm = CredentialsDialogViewModel.OfEmpty(); + CredentialsDialogFactory.CreateAndShow(_stateManager, cvm, null); + } + + private class FailureCollector { + private readonly ConnectionManagerDialog _cmDialog; + private readonly object _sync = new(); + private int _rowsLeft = 0; + private readonly List _failures = new(); + + public FailureCollector(ConnectionManagerDialog cmDialog, int rowsLeft) { + _cmDialog = cmDialog; + _rowsLeft = rowsLeft; + } + + public void OnFailure(EndpointId id, string reason) { + lock (_sync) { + _failures.Add(reason); + } + + FinalSteps(); + } + + public void OnSuccess(EndpointId id) { + FinalSteps(); + } + + private void FinalSteps() { + string text; + lock (_sync) { + --_rowsLeft; + if (_rowsLeft > 0 || _failures.Count == 0) { + return; + } + + text = string.Join(Environment.NewLine, _failures); + } + + const string caption = "Couldn't delete some selections"; + _cmDialog.Invoke(() => { + var mbox = new DeephavenMessageBox(caption, text, false); + mbox.ShowDialog(_cmDialog); + }); + } + } + + void OnDeleteButtonClicked(ConnectionManagerDialogRow[] rows) { + if (_workerThread.EnqueueOrNop(() => OnDeleteButtonClicked(rows))) { + return; + } + + var fc = new FailureCollector(_cmDialog, rows.Length); + foreach (var row in rows) { + if (!_rowToManager.TryGetValue(row, out var manager)) { + continue; + } + manager.DoDelete(fc.OnSuccess, fc.OnFailure); + } + } + + void OnReconnectButtonClicked(ConnectionManagerDialogRow[] rows) { + if (_workerThread.EnqueueOrNop(() => OnReconnectButtonClicked(rows))) { + return; + } + + foreach (var row in rows) { + if (!_rowToManager.TryGetValue(row, out var manager)) { + continue; + } + manager.DoReconnect(); + } + } + + void OnMakeDefaultButtonClicked(ConnectionManagerDialogRow[] rows) { + if (_workerThread.EnqueueOrNop(() => OnMakeDefaultButtonClicked(rows))) { + return; + } + + // Make the last selected row the default + if (rows.Length == 0) { + return; + } + + var row = rows[^1]; + if (!_rowToManager.TryGetValue(row, out var manager)) { + return; + } + + manager.DoSetAsDefault(); + } + + void OnEditButtonClicked(ConnectionManagerDialogRow[] rows) { + if (_workerThread.EnqueueOrNop(() => OnEditButtonClicked(rows))) { + return; + } + + foreach (var row in rows) { + if (!_rowToManager.TryGetValue(row, out var manager)) { + continue; + } + manager.DoEdit(); + } + } +} diff --git a/csharp/ExcelAddIn/managers/ConnectionManagerDialogRowManager.cs b/csharp/ExcelAddIn/managers/ConnectionManagerDialogRowManager.cs new file mode 100644 index 00000000000..1b98da60e65 --- /dev/null +++ b/csharp/ExcelAddIn/managers/ConnectionManagerDialogRowManager.cs @@ -0,0 +1,132 @@ +using Deephaven.ExcelAddIn.Factories; +using Deephaven.ExcelAddIn.Models; +using Deephaven.ExcelAddIn.Util; +using Deephaven.ExcelAddIn.Viewmodels; +using Deephaven.ExcelAddIn.ViewModels; + +namespace Deephaven.ExcelAddIn.Managers; + +public sealed class ConnectionManagerDialogRowManager : + IObserver>, + IObserver>, + IObserver, + IDisposable { + + public static ConnectionManagerDialogRowManager Create(ConnectionManagerDialogRow row, + EndpointId endpointId, StateManager stateManager) { + var result = new ConnectionManagerDialogRowManager(row, endpointId, stateManager); + result.Resubscribe(); + return result; + } + + private readonly ConnectionManagerDialogRow _row; + private readonly EndpointId _endpointId; + private readonly StateManager _stateManager; + private readonly WorkerThread _workerThread; + private readonly List _disposables = new(); + + private ConnectionManagerDialogRowManager(ConnectionManagerDialogRow row, EndpointId endpointId, + StateManager stateManager) { + _row = row; + _endpointId = endpointId; + _stateManager = stateManager; + _workerThread = stateManager.WorkerThread; + } + + public void Dispose() { + Unsubscribe(); + } + + private void Resubscribe() { + if (_workerThread.EnqueueOrNop(Resubscribe)) { + return; + } + + if (_disposables.Count != 0) { + throw new Exception("State error: already subscribed"); + } + // We watch for session and credential state changes in our ID + var d1 = _stateManager.SubscribeToSession(_endpointId, this); + var d2 = _stateManager.SubscribeToCredentials(_endpointId, this); + var d3 = _stateManager.SubscribeToDefaultEndpointSelection(this); + _disposables.AddRange(new[] { d1, d2, d3 }); + } + + private void Unsubscribe() { + if (_workerThread.EnqueueOrNop(Unsubscribe)) { + return; + } + var temp = _disposables.ToArray(); + _disposables.Clear(); + + foreach (var disposable in temp) { + disposable.Dispose(); + } + } + + public void OnNext(StatusOr value) { + _row.SetCredentialsSynced(value); + } + + public void OnNext(StatusOr value) { + _row.SetSessionSynced(value); + } + + public void OnNext(EndpointId? value) { + _row.SetDefaultEndpointIdSynced(value); + } + + public void DoEdit() { + var creds = _row.GetCredentialsSynced(); + // If we have valid credentials, then make a populated viewmodel. + // If we don't, then make an empty viewmodel with only Id populated. + var cvm = creds.AcceptVisitor( + crs => CredentialsDialogViewModel.OfIdAndCredentials(_endpointId.Id, crs), + _ => CredentialsDialogViewModel.OfIdButOtherwiseEmpty(_endpointId.Id)); + CredentialsDialogFactory.CreateAndShow(_stateManager, cvm, _endpointId); + } + + public void DoDelete(Action onSuccess, Action onFailure) { + if (_workerThread.EnqueueOrNop(() => DoDelete(onSuccess, onFailure))) { + return; + } + + // Strategy: + // 1. Unsubscribe to everything + // 2. If it turns out that we were the last subscriber to the credentials, then great, the + // delete can proceed. + // 3. If the credentials we are deleting are the default credentials, then unset default credentials + // 4. Otherwise (there is some other subscriber to the credentials), then the delete operation + // should be denied. In that case we restore our state by resubscribing to everything. + Unsubscribe(); + _stateManager.TryDeleteCredentials(_endpointId, + () => onSuccess(_endpointId), + reason => { + Resubscribe(); + onFailure(_endpointId, reason); + }); + } + + public void DoReconnect() { + _stateManager.Reconnect(_endpointId); + } + + public void DoSetAsDefault() { + // If the connection is already the default, do nothing. + if (_row.IsDefault) { + return; + } + + _stateManager.SetDefaultEndpointId(_endpointId); + } + + public void OnCompleted() { + // TODO(kosak) + throw new NotImplementedException(); + } + + public void OnError(Exception error) { + // TODO(kosak) + throw new NotImplementedException(); + } +} diff --git a/csharp/ExcelAddIn/models/Session.cs b/csharp/ExcelAddIn/models/Session.cs index b3851ef7a77..5f3d34eb02a 100644 --- a/csharp/ExcelAddIn/models/Session.cs +++ b/csharp/ExcelAddIn/models/Session.cs @@ -26,7 +26,13 @@ public override T Visit(Func onCore, Func } public override void Dispose() { - Utility.Exchange(ref _client, null)?.Dispose(); + var temp = Utility.Exchange(ref _client, null); + if (temp == null) { + return; + } + + // Do the actual dispose work on a helper thread. + Utility.RunInBackground(temp.Dispose); } public Client Client { @@ -48,11 +54,17 @@ public override T Visit(Func onCore, Func } public override void Dispose() { - if (workerThread.InvokeIfRequired(Dispose)) { + if (workerThread.EnqueueOrNop(Dispose)) { + return; + } + + var temp = Utility.Exchange(ref _sessionManager, null); + if (temp == null) { return; } - Utility.Exchange(ref _sessionManager, null)?.Dispose(); + // Do the actual dispose work on a helper thread. + Utility.RunInBackground(temp.Dispose); } public SessionManager SessionManager { diff --git a/csharp/ExcelAddIn/models/SimpleModels.cs b/csharp/ExcelAddIn/models/SimpleModels.cs index 02a89b370b7..4593e58fef9 100644 --- a/csharp/ExcelAddIn/models/SimpleModels.cs +++ b/csharp/ExcelAddIn/models/SimpleModels.cs @@ -4,6 +4,10 @@ public record AddOrRemove(bool IsAdd, T Value) { public static AddOrRemove OfAdd(T value) { return new AddOrRemove(true, value); } + + public static AddOrRemove OfRemove(T value) { + return new AddOrRemove(false, value); + } } public record EndpointId(string Id) { diff --git a/csharp/ExcelAddIn/models/TableTriple.cs b/csharp/ExcelAddIn/models/TableTriple.cs index 95d7e7847b5..89a298dd350 100644 --- a/csharp/ExcelAddIn/models/TableTriple.cs +++ b/csharp/ExcelAddIn/models/TableTriple.cs @@ -1,5 +1,10 @@ namespace Deephaven.ExcelAddIn.Models; +public record PersistentQueryKey( + EndpointId EndpointId, + PersistentQueryId? PersistentQueryId) { +} + public record TableTriple( EndpointId? EndpointId, PersistentQueryId? PersistentQueryId, @@ -35,3 +40,10 @@ public static bool TryParse(string text, out TableTriple result, out string erro return true; } } + +public record TableQuad( + EndpointId? EndpointId, + PersistentQueryId? PersistentQueryId, + string TableName, + string Condition) { +} diff --git a/csharp/ExcelAddIn/operations/SnapshotOperation.cs b/csharp/ExcelAddIn/operations/SnapshotOperation.cs index 70b577de0e3..b9a299b8309 100644 --- a/csharp/ExcelAddIn/operations/SnapshotOperation.cs +++ b/csharp/ExcelAddIn/operations/SnapshotOperation.cs @@ -8,18 +8,15 @@ namespace Deephaven.ExcelAddIn.Operations; internal class SnapshotOperation : IExcelObservable, IObserver> { - private readonly TableTriple _tableDescriptor; - private readonly string _filter; + private readonly TableQuad _tableQuad; private readonly bool _wantHeaders; private readonly StateManager _stateManager; private readonly ObserverContainer> _observers = new(); private readonly WorkerThread _workerThread; private IDisposable? _filteredTableDisposer = null; - public SnapshotOperation(TableTriple tableDescriptor, string filter, bool wantHeaders, - StateManager stateManager) { - _tableDescriptor = tableDescriptor; - _filter = filter; + public SnapshotOperation(TableQuad tableQuad, bool wantHeaders, StateManager stateManager) { + _tableQuad = tableQuad; _wantHeaders = wantHeaders; _stateManager = stateManager; // Convenience @@ -28,42 +25,40 @@ public SnapshotOperation(TableTriple tableDescriptor, string filter, bool wantHe public IDisposable Subscribe(IExcelObserver observer) { var wrappedObserver = ExcelDnaHelpers.WrapExcelObserver(observer); - _workerThread.Invoke(() => { + _workerThread.EnqueueOrRun(() => { _observers.Add(wrappedObserver, out var isFirst); if (isFirst) { - _filteredTableDisposer = _stateManager.SubscribeToTableTriple(_tableDescriptor, _filter, this); + _filteredTableDisposer = _stateManager.SubscribeToTable(_tableQuad, this); } }); - return ActionAsDisposable.Create(() => { - _workerThread.Invoke(() => { - _observers.Remove(wrappedObserver, out var wasLast); - if (!wasLast) { - return; - } + return _workerThread.EnqueueOrRunWhenDisposed(() => { + _observers.Remove(wrappedObserver, out var wasLast); + if (!wasLast) { + return; + } - Utility.Exchange(ref _filteredTableDisposer, null)?.Dispose(); - }); + Utility.Exchange(ref _filteredTableDisposer, null)?.Dispose(); }); } - public void OnNext(StatusOr soth) { - if (_workerThread.InvokeIfRequired(() => OnNext(soth))) { + public void OnNext(StatusOr tableHandle) { + if (_workerThread.EnqueueOrNop(() => OnNext(tableHandle))) { return; } - if (!soth.GetValueOrStatus(out var tableHandle, out var status)) { + if (!tableHandle.GetValueOrStatus(out var th, out var status)) { _observers.SendStatus(status); return; } - _observers.SendStatus($"Snapshotting \"{_tableDescriptor.TableName}\""); + _observers.SendStatus($"Snapshotting \"{_tableQuad.TableName}\""); try { - using var ct = tableHandle.ToClientTable(); - var result = Renderer.Render(ct, _wantHeaders); - _observers.SendValue(result); + using var ct = th.ToClientTable(); + var rendered = Renderer.Render(ct, _wantHeaders); + _observers.SendValue(rendered); } catch (Exception ex) { _observers.SendStatus(ex.Message); } diff --git a/csharp/ExcelAddIn/operations/SubscribeOperation.cs b/csharp/ExcelAddIn/operations/SubscribeOperation.cs index e451861546a..f2478d67a53 100644 --- a/csharp/ExcelAddIn/operations/SubscribeOperation.cs +++ b/csharp/ExcelAddIn/operations/SubscribeOperation.cs @@ -2,27 +2,23 @@ using Deephaven.DeephavenClient.ExcelAddIn.Util; using Deephaven.ExcelAddIn.ExcelDna; using Deephaven.ExcelAddIn.Models; -using Deephaven.ExcelAddIn.Providers; using Deephaven.ExcelAddIn.Util; using ExcelDna.Integration; namespace Deephaven.ExcelAddIn.Operations; internal class SubscribeOperation : IExcelObservable, IObserver> { - private readonly TableTriple _tableDescriptor; - private readonly string _filter; + private readonly TableQuad _tableQuad; private readonly bool _wantHeaders; private readonly StateManager _stateManager; private readonly ObserverContainer> _observers = new(); private readonly WorkerThread _workerThread; - private IDisposable? _filteredTableDisposer = null; + private IDisposable? _tableDisposer = null; private TableHandle? _currentTableHandle = null; private SubscriptionHandle? _currentSubHandle = null; - public SubscribeOperation(TableTriple tableDescriptor, string filter, bool wantHeaders, - StateManager stateManager) { - _tableDescriptor = tableDescriptor; - _filter = filter; + public SubscribeOperation(TableQuad tableQuad, bool wantHeaders, StateManager stateManager) { + _tableQuad = tableQuad; _wantHeaders = wantHeaders; _stateManager = stateManager; // Convenience @@ -31,39 +27,40 @@ public SubscribeOperation(TableTriple tableDescriptor, string filter, bool wantH public IDisposable Subscribe(IExcelObserver observer) { var wrappedObserver = ExcelDnaHelpers.WrapExcelObserver(observer); - _workerThread.Invoke(() => { + _workerThread.EnqueueOrRun(() => { _observers.Add(wrappedObserver, out var isFirst); if (isFirst) { - _filteredTableDisposer = _stateManager.SubscribeToTableTriple(_tableDescriptor, _filter, this); + _tableDisposer = _stateManager.SubscribeToTable(_tableQuad, this); } }); return ActionAsDisposable.Create(() => { - _workerThread.Invoke(() => { + _workerThread.EnqueueOrRun(() => { _observers.Remove(wrappedObserver, out var wasLast); if (!wasLast) { return; } - var temp = _filteredTableDisposer; - _filteredTableDisposer = null; - temp?.Dispose(); + Utility.Exchange(ref _tableDisposer, null)?.Dispose(); }); }); } public void OnNext(StatusOr soth) { - if (_workerThread.InvokeIfRequired(() => OnNext(soth))) { + if (_workerThread.EnqueueOrNop(() => OnNext(soth))) { return; } // First tear down old state if (_currentTableHandle != null) { - _currentTableHandle.Unsubscribe(_currentSubHandle!); - _currentSubHandle!.Dispose(); + if (_currentSubHandle != null) { + _currentTableHandle.Unsubscribe(_currentSubHandle!); + _currentSubHandle!.Dispose(); + _currentSubHandle = null; + } + _currentTableHandle = null; - _currentSubHandle = null; } if (!soth.GetValueOrStatus(out var tableHandle, out var status)) { @@ -71,7 +68,7 @@ public void OnNext(StatusOr soth) { return; } - _observers.SendStatus($"Subscribing to \"{_tableDescriptor.TableName}\""); + _observers.SendStatus($"Subscribing to \"{_tableQuad.TableName}\""); _currentTableHandle = tableHandle; _currentSubHandle = _currentTableHandle.Subscribe(new MyTickingCallback(_observers, _wantHeaders)); diff --git a/csharp/ExcelAddIn/providers/ClientProvider.cs b/csharp/ExcelAddIn/providers/ClientProvider.cs deleted file mode 100644 index 5d296e35221..00000000000 --- a/csharp/ExcelAddIn/providers/ClientProvider.cs +++ /dev/null @@ -1,129 +0,0 @@ -using Deephaven.DeephavenClient; -using Deephaven.ExcelAddIn.Models; -using Deephaven.ExcelAddIn.Util; -using Deephaven.DeephavenClient.ExcelAddIn.Util; -using Deephaven.DheClient.Session; - -namespace Deephaven.ExcelAddIn.Providers; - -internal class ClientProvider( - WorkerThread workerThread, - TableTriple descriptor) : IObserver>, IObservable>, IDisposable { - - private readonly ObserverContainer> _observers = new(); - private StatusOr _client = StatusOr.OfStatus("[No Client]"); - private DndClient? _ownedDndClient = null; - - public IDisposable Subscribe(IObserver> observer) { - // We need to run this on our worker thread because we want to protect - // access to our dictionary. - workerThread.Invoke(() => { - _observers.Add(observer, out _); - observer.OnNext(_client); - }); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - _observers.Remove(observer, out _); - }); - }); - } - - public void Dispose() { - DisposeClientState(); - } - - public void OnNext(StatusOr session) { - // Get onto the worker thread if we're not already on it. - if (workerThread.InvokeIfRequired(() => OnNext(session))) { - return; - } - - try { - // Dispose whatever state we had before. - DisposeClientState(); - - // If the new state is just a status message, make that our status and transmit to our observers - if (!session.GetValueOrStatus(out var sb, out var status)) { - _observers.SetAndSendStatus(ref _client, status); - return; - } - - var pqId = descriptor.PersistentQueryId; - - // New state is a Core or CorePlus Session. - _ = sb.Visit(coreSession => { - if (pqId != null) { - _observers.SetAndSendStatus(ref _client, "[PQ Id Not Valid for Community Core]"); - return Unit.Instance; - } - - // It's a Core session so we have our Client. - _observers.SetAndSendValue(ref _client, coreSession.Client); - return Unit.Instance; // Essentially a "void" value that is ignored. - }, corePlusSession => { - // It's a CorePlus session so subscribe us to its PQ observer for the appropriate PQ ID - // If no PQ id was provided, that's a problem - if (pqId == null) { - _observers.SetAndSendStatus(ref _client, "[PQ Id is Required]"); - return Unit.Instance; - } - - // Connect to the PQ on a separate thread - Utility.RunInBackground(() => ConnectToPq(corePlusSession.SessionManager, pqId)); - return Unit.Instance; - }); - } catch (Exception ex) { - _observers.SetAndSendStatus(ref _client, ex.Message); - } - } - - /// - /// This is executed on a separate thread because it might take a while. - /// - /// - /// - private void ConnectToPq(SessionManager sessionManager, PersistentQueryId pqId) { - StatusOr result; - DndClient? dndClient = null; - try { - dndClient = sessionManager.ConnectToPqByName(pqId.Id, false); - result = StatusOr.OfValue(dndClient); - } catch (Exception ex) { - result = StatusOr.OfStatus(ex.Message); - } - - // commit the results, but on the worker thread - workerThread.Invoke(() => { - // This should normally be null, but maybe there's a race. - var oldDndClient = Utility.Exchange(ref _ownedDndClient, dndClient); - _observers.SetAndSend(ref _client, result); - - // Yet another thread - if (oldDndClient != null) { - Utility.RunInBackground(() => Utility.IgnoreExceptions(() => oldDndClient.Dispose())); - } - }); - } - - private void DisposeClientState() { - // Get onto the worker thread if we're not already on it. - if (workerThread.InvokeIfRequired(DisposeClientState)) { - return; - } - - var oldClient = Utility.Exchange(ref _ownedDndClient, null); - if (oldClient != null) { - _observers.SetAndSendStatus(ref _client, "Disposing client"); - oldClient.Dispose(); - } - } - - public void OnCompleted() { - throw new NotImplementedException(); - } - - public void OnError(Exception error) { - throw new NotImplementedException(); - } -} diff --git a/csharp/ExcelAddIn/providers/CredentialsProvider.cs b/csharp/ExcelAddIn/providers/CredentialsProvider.cs new file mode 100644 index 00000000000..9a2d8c2deee --- /dev/null +++ b/csharp/ExcelAddIn/providers/CredentialsProvider.cs @@ -0,0 +1,37 @@ +using Deephaven.ExcelAddIn.Models; +using Deephaven.ExcelAddIn.Util; + +namespace Deephaven.ExcelAddIn.Providers; + +internal class CredentialsProvider : IObservable> { + private readonly WorkerThread _workerThread; + private readonly ObserverContainer> _observers = new(); + private StatusOr _credentials = StatusOr.OfStatus("[No Credentials]"); + + public CredentialsProvider(StateManager stateManager) { + _workerThread = stateManager.WorkerThread; + } + + public void Init() { + // Do nothing + } + + public IDisposable Subscribe(IObserver> observer) { + _workerThread.EnqueueOrRun(() => { + _observers.Add(observer, out _); + observer.OnNext(_credentials); + }); + + return _workerThread.EnqueueOrRunWhenDisposed(() => _observers.Remove(observer, out _)); + } + + public void SetCredentials(CredentialsBase newCredentials) { + _observers.SetAndSendValue(ref _credentials, newCredentials); + } + + public void Resend() { + _observers.OnNext(_credentials); + } + + public int ObserverCountUnsafe => _observers.Count; +} diff --git a/csharp/ExcelAddIn/providers/DefaultEndpointTableProvider.cs b/csharp/ExcelAddIn/providers/DefaultEndpointTableProvider.cs new file mode 100644 index 00000000000..26d1dab5f35 --- /dev/null +++ b/csharp/ExcelAddIn/providers/DefaultEndpointTableProvider.cs @@ -0,0 +1,83 @@ +using Deephaven.DeephavenClient; +using Deephaven.ExcelAddIn.Models; +using Deephaven.ExcelAddIn.Util; +using System.Diagnostics; + +namespace Deephaven.ExcelAddIn.Providers; + +internal class DefaultEndpointTableProvider : + IObserver>, + IObserver, + // IObservable>, // redundant, part of ITableProvider + ITableProvider { + private const string UnsetTableHandleText = "[No Default Connection]"; + + private readonly StateManager _stateManager; + private readonly PersistentQueryId? _persistentQueryId; + private readonly string _tableName; + private readonly string _condition; + private readonly WorkerThread _workerThread; + private Action? _onDispose; + private IDisposable? _endpointSubscriptionDisposer = null; + private IDisposable? _upstreamSubscriptionDisposer = null; + private readonly ObserverContainer> _observers = new(); + private StatusOr _tableHandle = StatusOr.OfStatus(UnsetTableHandleText); + + public DefaultEndpointTableProvider(StateManager stateManager, + PersistentQueryId? persistentQueryId, string tableName, string condition, + Action onDispose) { + _stateManager = stateManager; + _workerThread = stateManager.WorkerThread; + _persistentQueryId = persistentQueryId; + _tableName = tableName; + _condition = condition; + _onDispose = onDispose; + } + + public void Init() { + _endpointSubscriptionDisposer = _stateManager.SubscribeToDefaultEndpointSelection(this); + } + + public IDisposable Subscribe(IObserver> observer) { + _workerThread.EnqueueOrRun(() => { + _observers.Add(observer, out _); + observer.OnNext(_tableHandle); + }); + + return _workerThread.EnqueueOrRunWhenDisposed(() => { + _observers.Remove(observer, out var isLast); + if (!isLast) { + return; + } + + Utility.Exchange(ref _endpointSubscriptionDisposer, null)?.Dispose(); + Utility.Exchange(ref _onDispose, null)?.Invoke(); + }); + } + + public void OnNext(EndpointId? endpointId) { + // Unsubscribe from old upstream + Utility.Exchange(ref _upstreamSubscriptionDisposer, null)?.Dispose(); + + // If endpoint is null, then don't subscribe to anything. + if (endpointId == null) { + _observers.SetAndSendStatus(ref _tableHandle, UnsetTableHandleText); + return; + } + + var tq = new TableQuad(endpointId, _persistentQueryId, _tableName, _condition); + _upstreamSubscriptionDisposer = _stateManager.SubscribeToTable(tq, this); + } + + public void OnNext(StatusOr value) { + _observers.SetAndSend(ref _tableHandle, value); + } + + public void OnCompleted() { + throw new NotImplementedException(); + } + + public void OnError(Exception error) { + throw new NotImplementedException(); + } +} diff --git a/csharp/ExcelAddIn/providers/DefaultSessionProvider.cs b/csharp/ExcelAddIn/providers/DefaultSessionProvider.cs deleted file mode 100644 index f38f8bc8a8e..00000000000 --- a/csharp/ExcelAddIn/providers/DefaultSessionProvider.cs +++ /dev/null @@ -1,88 +0,0 @@ -using Deephaven.DeephavenClient.ExcelAddIn.Util; -using Deephaven.ExcelAddIn.Models; -using Deephaven.ExcelAddIn.Util; - -namespace Deephaven.ExcelAddIn.Providers; - -internal class DefaultSessionProvider(WorkerThread workerThread) : - IObserver>, IObserver>, - IObservable>, IObservable> { - private StatusOr _credentials = StatusOr.OfStatus("[Not set]"); - private StatusOr _session = StatusOr.OfStatus("[Not connected]"); - private readonly ObserverContainer> _credentialsObservers = new(); - private readonly ObserverContainer> _sessionObservers = new(); - private SessionProvider? _parent = null; - private IDisposable? _credentialsSubDisposer = null; - private IDisposable? _sessionSubDisposer = null; - - public IDisposable Subscribe(IObserver> observer) { - workerThread.Invoke(() => { - // New observer gets added to the collection and then notified of the current status. - _credentialsObservers.Add(observer, out _); - observer.OnNext(_credentials); - }); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - _credentialsObservers.Remove(observer, out _); - }); - }); - } - - public IDisposable Subscribe(IObserver> observer) { - workerThread.Invoke(() => { - // New observer gets added to the collection and then notified of the current status. - _sessionObservers.Add(observer, out _); - observer.OnNext(_session); - }); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - _sessionObservers.Remove(observer, out _); - }); - }); - } - - public void OnNext(StatusOr value) { - if (workerThread.InvokeIfRequired(() => OnNext(value))) { - return; - } - _credentials = value; - _credentialsObservers.OnNext(_credentials); - } - - public void OnNext(StatusOr value) { - if (workerThread.InvokeIfRequired(() => OnNext(value))) { - return; - } - _session = value; - _sessionObservers.OnNext(_session); - } - - public void OnCompleted() { - // TODO(kosak) - throw new NotImplementedException(); - } - - public void OnError(Exception error) { - // TODO(kosak) - throw new NotImplementedException(); - } - - public void SetParent(SessionProvider? newParent) { - if (workerThread.InvokeIfRequired(() => SetParent(newParent))) { - return; - } - - _parent = newParent; - Utility.Exchange(ref _credentialsSubDisposer, null)?.Dispose(); - Utility.Exchange(ref _sessionSubDisposer, null)?.Dispose(); - - if (_parent == null) { - return; - } - - _credentialsSubDisposer = _parent.Subscribe((IObserver>)this); - _sessionSubDisposer = _parent.Subscribe((IObserver>)this); - } -} diff --git a/csharp/ExcelAddIn/providers/FilteredTableProvider.cs b/csharp/ExcelAddIn/providers/FilteredTableProvider.cs new file mode 100644 index 00000000000..2774ec31c94 --- /dev/null +++ b/csharp/ExcelAddIn/providers/FilteredTableProvider.cs @@ -0,0 +1,106 @@ +using System.Diagnostics; +using Deephaven.DeephavenClient; +using Deephaven.ExcelAddIn.Models; +using Deephaven.ExcelAddIn.Util; + +namespace Deephaven.ExcelAddIn.Providers; + +internal class FilteredTableProvider : + IObserver>, + // IObservable>, // redundant, part of ITableProvider + ITableProvider { + + private readonly StateManager _stateManager; + private readonly WorkerThread _workerThread; + private readonly EndpointId _endpointId; + private readonly PersistentQueryId? _persistentQueryId; + private readonly string _tableName; + private readonly string _condition; + private Action? _onDispose; + private IDisposable? _tableHandleSubscriptionDisposer = null; + private readonly ObserverContainer> _observers = new(); + private StatusOr _filteredTableHandle = StatusOr.OfStatus("[No Filtered Table]"); + + public FilteredTableProvider(StateManager stateManager, + EndpointId endpointId, PersistentQueryId? persistentQueryId, string tableName, string condition, + Action onDispose) { + _stateManager = stateManager; + _workerThread = stateManager.WorkerThread; + _endpointId = endpointId; + _persistentQueryId = persistentQueryId; + _tableName = tableName; + _condition = condition; + _onDispose = onDispose; + } + + public void Init() { + // Subscribe to a condition-free table + var tq = new TableQuad(_endpointId, _persistentQueryId, _tableName, ""); + Debug.WriteLine($"FTP is subscribing to TableHandle with {tq}"); + _tableHandleSubscriptionDisposer = _stateManager.SubscribeToTable(tq, this); + } + + public IDisposable Subscribe(IObserver> observer) { + _workerThread.EnqueueOrRun(() => { + _observers.Add(observer, out _); + observer.OnNext(_filteredTableHandle); + }); + + return _workerThread.EnqueueOrRunWhenDisposed(() => { + _observers.Remove(observer, out var isLast); + if (!isLast) { + return; + } + + Utility.Exchange(ref _tableHandleSubscriptionDisposer, null)?.Dispose(); + Utility.Exchange(ref _onDispose, null)?.Invoke(); + DisposeTableHandleState(); + }); + } + + public void OnNext(StatusOr tableHandle) { + // Get onto the worker thread if we're not already on it. + if (_workerThread.EnqueueOrNop(() => OnNext(tableHandle))) { + return; + } + + DisposeTableHandleState(); + + // If the new state is just a status message, make that our state and transmit to our observers + if (!tableHandle.GetValueOrStatus(out var th, out var status)) { + _observers.SetAndSendStatus(ref _filteredTableHandle, status); + return; + } + + // It's a real TableHandle so start fetching the table. First notify our observers. + _observers.SetAndSendStatus(ref _filteredTableHandle, "Filtering"); + + try { + var filtered = th.Where(_condition); + _observers.SetAndSendValue(ref _filteredTableHandle, filtered); + } catch (Exception ex) { + _observers.SetAndSendStatus(ref _filteredTableHandle, ex.Message); + } + } + + private void DisposeTableHandleState() { + if (_workerThread.EnqueueOrNop(DisposeTableHandleState)) { + return; + } + + _ = _filteredTableHandle.GetValueOrStatus(out var oldTh, out _); + _observers.SetAndSendStatus(ref _filteredTableHandle, "Disposing TableHandle"); + + if (oldTh != null) { + Utility.RunInBackground(oldTh.Dispose); + } + } + + public void OnCompleted() { + throw new NotImplementedException(); + } + + public void OnError(Exception error) { + throw new NotImplementedException(); + } +} diff --git a/csharp/ExcelAddIn/providers/ITableProvider.cs b/csharp/ExcelAddIn/providers/ITableProvider.cs new file mode 100644 index 00000000000..f0af94f4caa --- /dev/null +++ b/csharp/ExcelAddIn/providers/ITableProvider.cs @@ -0,0 +1,11 @@ +using Deephaven.DeephavenClient; +using Deephaven.ExcelAddIn.Util; + +namespace Deephaven.ExcelAddIn.Providers; + +/// +/// Common interface for TableProvider, FilteredTableProvider, and DefaultEndpointTableProvider +/// +public interface ITableProvider : IObservable> { + void Init(); +} diff --git a/csharp/ExcelAddIn/providers/PersistentQueryProvider.cs b/csharp/ExcelAddIn/providers/PersistentQueryProvider.cs new file mode 100644 index 00000000000..11c240be9ff --- /dev/null +++ b/csharp/ExcelAddIn/providers/PersistentQueryProvider.cs @@ -0,0 +1,110 @@ +using Deephaven.DeephavenClient; +using Deephaven.ExcelAddIn.Models; +using Deephaven.ExcelAddIn.Util; + +namespace Deephaven.ExcelAddIn.Providers; + +internal class PersistentQueryProvider : + IObserver>, IObservable> { + + private readonly StateManager _stateManager; + private readonly WorkerThread _workerThread; + private readonly EndpointId _endpointId; + private readonly PersistentQueryId? _pqId; + private Action? _onDispose; + private IDisposable? _upstreamSubscriptionDisposer = null; + private readonly ObserverContainer> _observers = new(); + private StatusOr _client = StatusOr.OfStatus("[No Client]"); + private Client? _ownedDndClient = null; + + public PersistentQueryProvider(StateManager stateManager, + EndpointId endpointId, PersistentQueryId? pqId, Action onDispose) { + _stateManager = stateManager; + _workerThread = stateManager.WorkerThread; + _endpointId = endpointId; + _pqId = pqId; + _onDispose = onDispose; + } + + public void Init() { + _upstreamSubscriptionDisposer = _stateManager.SubscribeToSession(_endpointId, this); + } + + public IDisposable Subscribe(IObserver> observer) { + _workerThread.EnqueueOrRun(() => { + _observers.Add(observer, out _); + observer.OnNext(_client); + }); + + return _workerThread.EnqueueOrRunWhenDisposed(() => { + _observers.Remove(observer, out var isLast); + if (!isLast) { + return; + } + + Utility.Exchange(ref _upstreamSubscriptionDisposer, null)?.Dispose(); + Utility.Exchange(ref _onDispose, null)?.Invoke(); + DisposeClientState(); + }); + } + + public void OnNext(StatusOr sessionBase) { + if (_workerThread.EnqueueOrNop(() => OnNext(sessionBase))) { + return; + } + + DisposeClientState(); + + // If the new state is just a status message, make that our state and transmit to our observers + if (!sessionBase.GetValueOrStatus(out var sb, out var status)) { + _observers.SetAndSendStatus(ref _client, status); + return; + } + + // It's a real Session so start fetching it. Also do some validity checking on the PQ id. + _ = sb.Visit( + core => { + var result = _pqId == null + ? StatusOr.OfValue(core.Client) + : StatusOr.OfStatus("PQ specified, but Community Core cannot connect to a PQ"); + _observers.SetAndSend(ref _client, result); + return Unit.Instance; + }, + corePlus => { + if (_pqId == null) { + _observers.SetAndSendStatus(ref _client, "Enterprise Core+ requires a PQ to be specified"); + return Unit.Instance; + } + + _observers.SetAndSendStatus(ref _client, $"Attaching to \"{_pqId}\""); + + try { + _ownedDndClient = corePlus.SessionManager.ConnectToPqByName(_pqId.Id, false); + _observers.SetAndSendValue(ref _client, _ownedDndClient); + } catch (Exception ex) { + _observers.SetAndSendStatus(ref _client, ex.Message); + } + return Unit.Instance; + }); + } + + private void DisposeClientState() { + if (_workerThread.EnqueueOrNop(DisposeClientState)) { + return; + } + + _observers.SetAndSendStatus(ref _client, "Disposing Client"); + var oldClient = Utility.Exchange(ref _ownedDndClient, null); + if (oldClient != null) { + Utility.RunInBackground(oldClient.Dispose); + } + } + + public void OnCompleted() { + throw new NotImplementedException(); + } + + public void OnError(Exception error) { + throw new NotImplementedException(); + } +} diff --git a/csharp/ExcelAddIn/providers/SessionProvider.cs b/csharp/ExcelAddIn/providers/SessionProvider.cs index 11882df8144..f34fefdb6d8 100644 --- a/csharp/ExcelAddIn/providers/SessionProvider.cs +++ b/csharp/ExcelAddIn/providers/SessionProvider.cs @@ -1,139 +1,101 @@ -using Deephaven.DeephavenClient.ExcelAddIn.Util; -using Deephaven.ExcelAddIn.Factories; +using Deephaven.ExcelAddIn.Factories; using Deephaven.ExcelAddIn.Models; using Deephaven.ExcelAddIn.Util; -using System.Net; namespace Deephaven.ExcelAddIn.Providers; -internal class SessionProvider(WorkerThread workerThread) : IObservable>, IObservable>, IDisposable { - private StatusOr _credentials = StatusOr.OfStatus("[Not set]"); +internal class SessionProvider : IObserver>, IObservable> { + private readonly StateManager _stateManager; + private readonly WorkerThread _workerThread; + private readonly EndpointId _endpointId; + private Action? _onDispose; + private IDisposable? _upstreamSubscriptionDisposer = null; private StatusOr _session = StatusOr.OfStatus("[Not connected]"); - private readonly ObserverContainer> _credentialsObservers = new(); - private readonly ObserverContainer> _sessionObservers = new(); - /// - /// This is used to ignore the results from multiple invocations of "SetCredentials". - /// - private readonly SimpleAtomicReference _sharedSetCredentialsCookie = new(new object()); - - public void Dispose() { - // Get on the worker thread if not there already. - if (workerThread.InvokeIfRequired(Dispose)) { - return; - } - - // TODO(kosak) - // I feel like we should send an OnComplete to any remaining observers - - if (!_session.GetValueOrStatus(out var sess, out _)) { - return; - } - - _sessionObservers.SetAndSendStatus(ref _session, "Disposing"); - sess.Dispose(); + private readonly ObserverContainer> _observers = new(); + private readonly VersionTracker _versionTracker = new(); + + public SessionProvider(StateManager stateManager, EndpointId endpointId, Action onDispose) { + _stateManager = stateManager; + _workerThread = stateManager.WorkerThread; + _endpointId = endpointId; + _onDispose = onDispose; } - /// - /// Subscribe to credentials changes - /// - /// - /// - public IDisposable Subscribe(IObserver> observer) { - workerThread.Invoke(() => { - // New observer gets added to the collection and then notified of the current status. - _credentialsObservers.Add(observer, out _); - observer.OnNext(_credentials); - }); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - _credentialsObservers.Remove(observer, out _); - }); - }); + public void Init() { + _upstreamSubscriptionDisposer = _stateManager.SubscribeToCredentials(_endpointId, this); } /// /// Subscribe to session changes /// - /// - /// public IDisposable Subscribe(IObserver> observer) { - workerThread.Invoke(() => { - // New observer gets added to the collection and then notified of the current status. - _sessionObservers.Add(observer, out _); + _workerThread.EnqueueOrRun(() => { + _observers.Add(observer, out _); observer.OnNext(_session); }); - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - _sessionObservers.Remove(observer, out _); - }); + return _workerThread.EnqueueOrRunWhenDisposed(() => { + _observers.Remove(observer, out var isLast); + if (!isLast) { + return; + } + + Utility.Exchange(ref _upstreamSubscriptionDisposer, null)?.Dispose(); + Utility.Exchange(ref _onDispose, null)?.Invoke(); + DisposeSessionState(); }); } - public void SetCredentials(CredentialsBase credentials) { - // Get on the worker thread if not there already. - if (workerThread.InvokeIfRequired(() => SetCredentials(credentials))) { + public void OnNext(StatusOr credentials) { + if (_workerThread.EnqueueOrNop(() => OnNext(credentials))) { return; } - // Dispose existing session - if (_session.GetValueOrStatus(out var sess, out _)) { - _sessionObservers.SetAndSendStatus(ref _session, "Disposing session"); - sess.Dispose(); - } + DisposeSessionState(); - _credentialsObservers.SetAndSendValue(ref _credentials, credentials); + if (!credentials.GetValueOrStatus(out var cbase, out var status)) { + _observers.SetAndSendStatus(ref _session, status); + return; + } - _sessionObservers.SetAndSendStatus(ref _session, "Trying to connect"); + _observers.SetAndSendStatus(ref _session, "Trying to connect"); - Utility.RunInBackground(() => CreateSessionBaseInSeparateThread(credentials)); + var cookie = _versionTracker.SetNewVersion(); + Utility.RunInBackground(() => CreateSessionBaseInSeparateThread(cbase, cookie)); } - void CreateSessionBaseInSeparateThread(CredentialsBase credentials) { - // Make a unique sentinel object to indicate that this thread should be - // the one privileged to provide the system with the Session corresponding - // to the credentials. If SetCredentials isn't called in the meantime, - // we will go ahead and provide our answer to the system. However, if - // SetCredentials is called again, triggering a new thread, then that - // new thread will usurp our privilege and it will be the one to provide - // the answer. - var localLatestCookie = new object(); - _sharedSetCredentialsCookie.Value = localLatestCookie; - + private void CreateSessionBaseInSeparateThread(CredentialsBase credentials, VersionTrackerCookie versionCookie) { + SessionBase? sb = null; StatusOr result; try { // This operation might take some time. - var sb = SessionBaseFactory.Create(credentials, workerThread); + sb = SessionBaseFactory.Create(credentials, _workerThread); result = StatusOr.OfValue(sb); } catch (Exception ex) { result = StatusOr.OfStatus(ex.Message); } - // If sharedTestCredentialsCookie is still the same, then our privilege - // has not been usurped and we can provide our answer to the system. - // On the other hand, if it has changed, then we will just throw away our work. - if (!ReferenceEquals(localLatestCookie, _sharedSetCredentialsCookie.Value)) { - // Our results are moot. Dispose of them. - if (result.GetValueOrStatus(out var sb, out _)) { - sb.Dispose(); - } + // Some time has passed. It's possible that the VersionTracker has been reset + // with a newer version. If so, we should throw away our work and leave. + if (!versionCookie.IsCurrent) { + sb?.Dispose(); return; } // Our results are valid. Keep them and tell everyone about it (on the worker thread). - workerThread.Invoke(() => _sessionObservers.SetAndSend(ref _session, result)); + _workerThread.EnqueueOrRun(() => _observers.SetAndSend(ref _session, result)); } - public void Reconnect() { - // Get on the worker thread if not there already. - if (workerThread.InvokeIfRequired(Reconnect)) { + private void DisposeSessionState() { + if (_workerThread.EnqueueOrNop(DisposeSessionState)) { return; } - // We implement this as a SetCredentials call, with credentials we already have. - if (_credentials.GetValueOrStatus(out var creds, out _)) { - SetCredentials(creds); + _ = _session.GetValueOrStatus(out var oldSession, out _); + _observers.SetAndSendStatus(ref _session, "Disposing Session"); + + if (oldSession != null) { + Utility.RunInBackground(oldSession.Dispose); } } diff --git a/csharp/ExcelAddIn/providers/SessionProviders.cs b/csharp/ExcelAddIn/providers/SessionProviders.cs deleted file mode 100644 index 5e5db2366e3..00000000000 --- a/csharp/ExcelAddIn/providers/SessionProviders.cs +++ /dev/null @@ -1,108 +0,0 @@ -using Deephaven.DeephavenClient.ExcelAddIn.Util; -using Deephaven.ExcelAddIn.Models; -using Deephaven.ExcelAddIn.Util; - -namespace Deephaven.ExcelAddIn.Providers; - -internal class SessionProviders(WorkerThread workerThread) : IObservable> { - private readonly DefaultSessionProvider _defaultProvider = new(workerThread); - private readonly Dictionary _providerMap = new(); - private readonly ObserverContainer> _endpointsObservers = new(); - - public IDisposable Subscribe(IObserver> observer) { - IDisposable? disposable = null; - // We need to run this on our worker thread because we want to protect - // access to our dictionary. - workerThread.Invoke(() => { - _endpointsObservers.Add(observer, out _); - // To avoid any further possibility of reentrancy while iterating over the dict, - // make a copy of the keys - var keys = _providerMap.Keys.ToArray(); - foreach (var endpointId in keys) { - observer.OnNext(AddOrRemove.OfAdd(endpointId)); - } - }); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - Utility.Exchange(ref disposable, null)?.Dispose(); - }); - }); - } - - public IDisposable SubscribeToSession(EndpointId id, IObserver> observer) { - IDisposable? disposable = null; - ApplyTo(id, sp => disposable = sp.Subscribe(observer)); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - Utility.Exchange(ref disposable, null)?.Dispose(); - }); - }); - } - - public IDisposable SubscribeToCredentials(EndpointId id, IObserver> observer) { - IDisposable? disposable = null; - ApplyTo(id, sp => disposable = sp.Subscribe(observer)); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - Utility.Exchange(ref disposable, null)?.Dispose(); - }); - }); - } - - public IDisposable SubscribeToDefaultSession(IObserver> observer) { - IDisposable? disposable = null; - workerThread.Invoke(() => { - disposable = _defaultProvider.Subscribe(observer); - }); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - Utility.Exchange(ref disposable, null)?.Dispose(); - }); - }); - } - - public IDisposable SubscribeToDefaultCredentials(IObserver> observer) { - IDisposable? disposable = null; - workerThread.Invoke(() => { - disposable = _defaultProvider.Subscribe(observer); - }); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - Utility.Exchange(ref disposable, null)?.Dispose(); - }); - }); - } - - public void SetCredentials(CredentialsBase credentials) { - ApplyTo(credentials.Id, sp => { - sp.SetCredentials(credentials); - }); - } - - public void SetDefaultCredentials(CredentialsBase credentials) { - ApplyTo(credentials.Id, _defaultProvider.SetParent); - } - - public void Reconnect(EndpointId id) { - ApplyTo(id, sp => sp.Reconnect()); - } - - private void ApplyTo(EndpointId id, Action action) { - if (workerThread.InvokeIfRequired(() => ApplyTo(id, action))) { - return; - } - - if (!_providerMap.TryGetValue(id, out var sp)) { - sp = new SessionProvider(workerThread); - _providerMap.Add(id, sp); - _endpointsObservers.OnNext(AddOrRemove.OfAdd(id)); - } - - action(sp); - } -} diff --git a/csharp/ExcelAddIn/providers/TableHandleProvider.cs b/csharp/ExcelAddIn/providers/TableHandleProvider.cs deleted file mode 100644 index db7ddccd4e9..00000000000 --- a/csharp/ExcelAddIn/providers/TableHandleProvider.cs +++ /dev/null @@ -1,98 +0,0 @@ -using Deephaven.DeephavenClient; -using Deephaven.ExcelAddIn.Models; -using Deephaven.ExcelAddIn.Util; -using Deephaven.DeephavenClient.ExcelAddIn.Util; - -namespace Deephaven.ExcelAddIn.Providers; - -internal class TableHandleProvider( - WorkerThread workerThread, - TableTriple descriptor, - string filter) : IObserver>, IObservable>, IDisposable { - - private readonly ObserverContainer> _observers = new(); - private StatusOr _tableHandle = StatusOr.OfStatus("[no TableHandle]"); - - public IDisposable Subscribe(IObserver> observer) { - // We need to run this on our worker thread because we want to protect - // access to our dictionary. - workerThread.Invoke(() => { - _observers.Add(observer, out _); - observer.OnNext(_tableHandle); - }); - - return ActionAsDisposable.Create(() => { - workerThread.Invoke(() => { - _observers.Remove(observer, out _); - }); - }); - } - - public void Dispose() { - // Get onto the worker thread if we're not already on it. - if (workerThread.InvokeIfRequired(Dispose)) { - return; - } - - DisposePqAndThState(); - } - - public void OnNext(StatusOr client) { - // Get onto the worker thread if we're not already on it. - if (workerThread.InvokeIfRequired(() => OnNext(client))) { - return; - } - - try { - // Dispose whatever state we had before. - DisposePqAndThState(); - - // If the new state is just a status message, make that our state and transmit to our observers - if (!client.GetValueOrStatus(out var cli, out var status)) { - _observers.SetAndSendStatus(ref _tableHandle, status); - return; - } - - // It's a real client so start fetching the table. First notify our observers. - _observers.SetAndSendStatus(ref _tableHandle, $"Fetching \"{descriptor.TableName}\""); - - // Now fetch the table. This might block but we're on the worker thread. In the future - // we might move this to yet another thread. - var th = cli.Manager.FetchTable(descriptor.TableName); - if (filter != "") { - // If there's a filter, take this table handle and surround it with a Where. - var temp = th; - th = temp.Where(filter); - temp.Dispose(); - } - - // Success! Make this our state and send the table handle to our observers. - _observers.SetAndSendValue(ref _tableHandle, th); - } catch (Exception ex) { - // Some exception. Make the exception message our state and send it to our observers. - _observers.SetAndSendStatus(ref _tableHandle, ex.Message); - } - } - - private void DisposePqAndThState() { - // Get onto the worker thread if we're not already on it. - if (workerThread.InvokeIfRequired(DisposePqAndThState)) { - return; - } - - _ = _tableHandle.GetValueOrStatus(out var oldTh, out _); - - if (oldTh != null) { - _observers.SetAndSendStatus(ref _tableHandle, "Disposing TableHandle"); - oldTh.Dispose(); - } - } - - public void OnCompleted() { - throw new NotImplementedException(); - } - - public void OnError(Exception error) { - throw new NotImplementedException(); - } -} diff --git a/csharp/ExcelAddIn/providers/TableProvider.cs b/csharp/ExcelAddIn/providers/TableProvider.cs new file mode 100644 index 00000000000..d4da46fef3a --- /dev/null +++ b/csharp/ExcelAddIn/providers/TableProvider.cs @@ -0,0 +1,100 @@ +using Deephaven.DeephavenClient; +using Deephaven.ExcelAddIn.Models; +using Deephaven.ExcelAddIn.Util; + +namespace Deephaven.ExcelAddIn.Providers; + +internal class TableProvider : + IObserver>, + // IObservable>, // redundant, part of ITableProvider + ITableProvider { + private const string UnsetTableHandleText = "[No Table]"; + + private readonly StateManager _stateManager; + private readonly WorkerThread _workerThread; + private readonly EndpointId _endpointId; + private readonly PersistentQueryId? _persistentQueryId; + private readonly string _tableName; + private Action? _onDispose; + private IDisposable? _pqSubscriptionDisposer = null; + private readonly ObserverContainer> _observers = new(); + private StatusOr _tableHandle = StatusOr.OfStatus(UnsetTableHandleText); + + public TableProvider(StateManager stateManager, EndpointId endpointId, + PersistentQueryId? persistentQueryId, string tableName, Action onDispose) { + _stateManager = stateManager; + _workerThread = stateManager.WorkerThread; + _endpointId = endpointId; + _persistentQueryId = persistentQueryId; + _tableName = tableName; + _onDispose = onDispose; + } + + public void Init() { + _pqSubscriptionDisposer = _stateManager.SubscribeToPersistentQuery( + _endpointId, _persistentQueryId, this); + } + + public IDisposable Subscribe(IObserver> observer) { + _workerThread.EnqueueOrRun(() => { + _observers.Add(observer, out _); + observer.OnNext(_tableHandle); + }); + + return _workerThread.EnqueueOrRunWhenDisposed(() => { + _observers.Remove(observer, out var isLast); + if (!isLast) { + return; + } + + Utility.Exchange(ref _pqSubscriptionDisposer, null)?.Dispose(); + Utility.Exchange(ref _onDispose, null)?.Invoke(); + DisposeTableHandleState(); + }); + } + + public void OnNext(StatusOr client) { + if (_workerThread.EnqueueOrNop(() => OnNext(client))) { + return; + } + + DisposeTableHandleState(); + + // If the new state is just a status message, make that our state and transmit to our observers + if (!client.GetValueOrStatus(out var cli, out var status)) { + _observers.SetAndSendStatus(ref _tableHandle, status); + return; + } + + // It's a real client so start fetching the table. First notify our observers. + _observers.SetAndSendStatus(ref _tableHandle, $"Fetching \"{_tableName}\""); + + try { + var th = cli.Manager.FetchTable(_tableName); + _observers.SetAndSendValue(ref _tableHandle, th); + } catch (Exception ex) { + _observers.SetAndSendStatus(ref _tableHandle, ex.Message); + } + } + + private void DisposeTableHandleState() { + if (_workerThread.EnqueueOrNop(DisposeTableHandleState)) { + return; + } + + _ = _tableHandle.GetValueOrStatus(out var oldTh, out _); + _observers.SetAndSendStatus(ref _tableHandle, UnsetTableHandleText); + + if (oldTh != null) { + Utility.RunInBackground(oldTh.Dispose); + } + } + + public void OnCompleted() { + throw new NotImplementedException(); + } + + public void OnError(Exception error) { + throw new NotImplementedException(); + } +} diff --git a/csharp/ExcelAddIn/util/ObservableConverter.cs b/csharp/ExcelAddIn/util/ObservableConverter.cs new file mode 100644 index 00000000000..b31defc40ee --- /dev/null +++ b/csharp/ExcelAddIn/util/ObservableConverter.cs @@ -0,0 +1,44 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using System.Windows.Forms; +using Deephaven.DeephavenClient.ExcelAddIn.Util; + +namespace Deephaven.ExcelAddIn.Util; + +internal static class ObservableConverter { + public static ObservableConverter Create( + Func converter, WorkerThread workerThread) { + return new ObservableConverter(converter, workerThread); + } +} + +internal class ObservableConverter(Func converter, WorkerThread workerThread) : + IObserver, IObservable { + private readonly ObserverContainer _observers = new(); + + public void OnNext(TFrom value) { + var converted = converter(value); + _observers.OnNext(converted); + } + + public void OnCompleted() { + _observers.OnCompleted(); + } + + public void OnError(Exception error) { + _observers.OnError(error); + } + + public IDisposable Subscribe(IObserver observer) { + workerThread.EnqueueOrRun(() => _observers.Add(observer, out _)); + + return ActionAsDisposable.Create(() => { + workerThread.EnqueueOrRun(() => { + _observers.Remove(observer, out _); + }); + }); + } +} diff --git a/csharp/ExcelAddIn/util/SimpleAtomicReference.cs b/csharp/ExcelAddIn/util/SimpleAtomicReference.cs deleted file mode 100644 index d5a2f76cd5b..00000000000 --- a/csharp/ExcelAddIn/util/SimpleAtomicReference.cs +++ /dev/null @@ -1,19 +0,0 @@ -namespace Deephaven.ExcelAddIn.Util; - -internal class SimpleAtomicReference(T value) { - private readonly object _sync = new(); - private T _value = value; - - public T Value { - get { - lock (_sync) { - return _value; - } - } - set { - lock (_sync) { - _value = value; - } - } - } -} diff --git a/csharp/ExcelAddIn/util/TableDescriptor.cs b/csharp/ExcelAddIn/util/TableDescriptor.cs deleted file mode 100644 index caa52c74e2a..00000000000 --- a/csharp/ExcelAddIn/util/TableDescriptor.cs +++ /dev/null @@ -1,2 +0,0 @@ -namespace Deephaven.ExcelAddIn.Util; - diff --git a/csharp/ExcelAddIn/util/Utility.cs b/csharp/ExcelAddIn/util/Utility.cs index 96e910b43c9..4b9de77ca3a 100644 --- a/csharp/ExcelAddIn/util/Utility.cs +++ b/csharp/ExcelAddIn/util/Utility.cs @@ -1,4 +1,6 @@  +using System.Diagnostics; + namespace Deephaven.ExcelAddIn.Util; internal static class Utility { @@ -9,21 +11,35 @@ public static T Exchange(ref T item, T newValue) { } public static void RunInBackground(Action a) { - new Thread(() => a()) { IsBackground = true }.Start(); - } - - public static void IgnoreExceptions(Action action) { - try { - action(); - } catch { - // Ignore errors + void Doit() { + try { + a(); + } catch (Exception e) { + Debug.WriteLine($"Ignoring exception {e}"); + } } + new Thread(Doit) { IsBackground = true }.Start(); } } public class Unit { - public static readonly Unit Instance = new Unit(); + public static readonly Unit Instance = new (); private Unit() { } } + +public class ValueHolder where T : class { + private T? _value = null; + + public T Value { + get { + if (_value == null) { + throw new Exception("Value is unset"); + } + + return _value; + } + set => _value = value; + } +} diff --git a/csharp/ExcelAddIn/util/VersionTracker.cs b/csharp/ExcelAddIn/util/VersionTracker.cs new file mode 100644 index 00000000000..a0fcbe48c71 --- /dev/null +++ b/csharp/ExcelAddIn/util/VersionTracker.cs @@ -0,0 +1,27 @@ +namespace Deephaven.ExcelAddIn.Util; + +internal class VersionTracker { + private readonly object _sync = new(); + private VersionTrackerCookie _cookie; + + public VersionTracker() { + _cookie = new VersionTrackerCookie(this); + } + + public VersionTrackerCookie SetNewVersion() { + lock (_sync) { + _cookie = new VersionTrackerCookie(this); + return _cookie; + } + } + + public bool HasCookie(VersionTrackerCookie cookie) { + lock (_sync) { + return ReferenceEquals(_cookie, cookie); + } + } +} + +internal class VersionTrackerCookie(VersionTracker owner) { + public bool IsCurrent => owner.HasCookie(this); +} diff --git a/csharp/ExcelAddIn/util/WorkerThread.cs b/csharp/ExcelAddIn/util/WorkerThread.cs index dc1d2858754..eceea724114 100644 --- a/csharp/ExcelAddIn/util/WorkerThread.cs +++ b/csharp/ExcelAddIn/util/WorkerThread.cs @@ -1,4 +1,5 @@ using System.Diagnostics; +using Deephaven.DeephavenClient.ExcelAddIn.Util; namespace Deephaven.ExcelAddIn.Util; @@ -18,13 +19,15 @@ public static WorkerThread Create() { private WorkerThread() { } - public void Invoke(Action action) { - if (!InvokeIfRequired(action)) { + // enquee or run + public void EnqueueOrRun(Action action) { + if (!EnqueueOrNop(action)) { action(); } } - public bool InvokeIfRequired(Action action) { + // conditionalenqueue + public bool EnqueueOrNop(Action action) { if (ReferenceEquals(Thread.CurrentThread, _thisThread)) { // Appending to thread queue was not required. Return false. return false; @@ -43,6 +46,10 @@ public bool InvokeIfRequired(Action action) { return true; } + public IDisposable EnqueueOrRunWhenDisposed(Action action) { + return ActionAsDisposable.Create(() => EnqueueOrRun(action)); + } + private void Doit() { while (true) { Action action; diff --git a/csharp/ExcelAddIn/viewmodels/ConnectionManagerDialogRow.cs b/csharp/ExcelAddIn/viewmodels/ConnectionManagerDialogRow.cs index 0e544131282..a34b9a6fbdb 100644 --- a/csharp/ExcelAddIn/viewmodels/ConnectionManagerDialogRow.cs +++ b/csharp/ExcelAddIn/viewmodels/ConnectionManagerDialogRow.cs @@ -1,21 +1,19 @@ -using Deephaven.ExcelAddIn.Factories; -using Deephaven.ExcelAddIn.ViewModels; -using System.ComponentModel; +using System.ComponentModel; using Deephaven.ExcelAddIn.Models; using Deephaven.ExcelAddIn.Util; namespace Deephaven.ExcelAddIn.Viewmodels; -public sealed class ConnectionManagerDialogRow(string id, StateManager stateManager) : - IObserver>, IObserver>, - INotifyPropertyChanged { +public sealed class ConnectionManagerDialogRow(string id) : INotifyPropertyChanged { + public event PropertyChangedEventHandler? PropertyChanged; private readonly object _sync = new(); private StatusOr _credentials = StatusOr.OfStatus("[Not set]"); private StatusOr _session = StatusOr.OfStatus("[Not connected]"); - private StatusOr _defaultCredentials = StatusOr.OfStatus("[Not set]"); + private EndpointId? _defaultEndpointId = null; + [DisplayName("Name")] public string Id { get; init; } = id; public string Status { @@ -28,6 +26,7 @@ public string Status { } } + [DisplayName("Server Type")] public string ServerType { get { var creds = GetCredentialsSynced(); @@ -41,40 +40,22 @@ public string ServerType { } } - public bool IsDefault => - _credentials.GetValueOrStatus(out var creds1, out _) && - _defaultCredentials.GetValueOrStatus(out var creds2, out _) && - creds1.Id == creds2.Id; - - public void SettingsClicked() { - var creds = GetCredentialsSynced(); - // If we have valid credentials, - var cvm = creds.AcceptVisitor( - crs => CredentialsDialogViewModel.OfIdAndCredentials(Id, crs), - _ => CredentialsDialogViewModel.OfIdButOtherwiseEmpty(Id)); - var cd = CredentialsDialogFactory.Create(stateManager, cvm); - cd.Show(); - } - - public void ReconnectClicked() { - stateManager.Reconnect(new EndpointId(Id)); - } - - public void IsDefaultClicked() { - // If the box is already checked, do nothing. - if (IsDefault) { - return; + [DisplayName("Default")] + public bool IsDefault { + get { + var id = Id; // readonly so no synchronization needed. + var defaultEp = GetDefaultEndpointIdSynced(); + return defaultEp != null && defaultEp.Id == id; } + } - // If we don't have credentials, then we can't make them the default. - if (!_credentials.GetValueOrStatus(out var creds, out _)) { - return; + public StatusOr GetCredentialsSynced() { + lock (_sync) { + return _credentials; } - - stateManager.SetDefaultCredentials(creds); } - public void OnNext(StatusOr value) { + public void SetCredentialsSynced(StatusOr value) { lock (_sync) { _credentials = value; } @@ -83,41 +64,30 @@ public void OnNext(StatusOr value) { OnPropertyChanged(nameof(IsDefault)); } - public void OnNext(StatusOr value) { + public EndpointId? GetDefaultEndpointIdSynced() { lock (_sync) { - _session = value; + return _defaultEndpointId; } - - OnPropertyChanged(nameof(Status)); } - public void SetDefaultCredentials(StatusOr creds) { + public void SetDefaultEndpointIdSynced(EndpointId? value) { lock (_sync) { - _defaultCredentials = creds; + _defaultEndpointId = value; } OnPropertyChanged(nameof(IsDefault)); } - public void OnCompleted() { - // TODO(kosak) - throw new NotImplementedException(); - } - - public void OnError(Exception error) { - // TODO(kosak) - throw new NotImplementedException(); - } - - private StatusOr GetCredentialsSynced() { + public StatusOr GetSessionSynced() { lock (_sync) { - return _credentials; + return _session; } } - private StatusOr GetSessionSynced() { + public void SetSessionSynced(StatusOr value) { lock (_sync) { - return _session; + _session = value; } + OnPropertyChanged(nameof(Status)); } private void OnPropertyChanged(string name) { diff --git a/csharp/ExcelAddIn/viewmodels/CredentialsDialogViewModel.cs b/csharp/ExcelAddIn/viewmodels/CredentialsDialogViewModel.cs index 6634640031f..86cae1c39e0 100644 --- a/csharp/ExcelAddIn/viewmodels/CredentialsDialogViewModel.cs +++ b/csharp/ExcelAddIn/viewmodels/CredentialsDialogViewModel.cs @@ -31,7 +31,7 @@ public static CredentialsDialogViewModel OfIdAndCredentials(string id, Credentia result.JsonUrl = corePlus.JsonUrl; result.UserId = corePlus.User; result.Password = corePlus.Password; - result.OperateAs = corePlus.OperateAs; + result.OperateAs = corePlus.OperateAs.Equals(corePlus.User) ? "" : corePlus.OperateAs; result.ValidateCertificate = corePlus.ValidateCertificate; return Unit.Instance; }); @@ -68,26 +68,25 @@ void CheckMissing(string field, string name) { } } - CheckMissing(_id, "Connection Id"); + CheckMissing(Id, "Connection Id"); if (!_isCorePlus) { - CheckMissing(_connectionString, "Connection String"); + CheckMissing(ConnectionString, "Connection String"); } else { - CheckMissing(_jsonUrl, "JSON URL"); - CheckMissing(_userId, "User Id"); - CheckMissing(_password, "Password"); - CheckMissing(_operateAs, "Operate As"); + CheckMissing(JsonUrl, "JSON URL"); + CheckMissing(UserId, "User Id"); + CheckMissing(Password, "Password"); } if (missingFields.Count > 0) { - errorText = string.Join(", ", missingFields); + errorText = string.Join(Environment.NewLine, missingFields); return false; } var epId = new EndpointId(_id); result = _isCorePlus - ? CredentialsBase.OfCorePlus(epId, _jsonUrl, _userId, _password, _operateAs, _validateCertificate) - : CredentialsBase.OfCore(epId, _connectionString, _sessionTypeIsPython); + ? CredentialsBase.OfCorePlus(epId, JsonUrl, UserId, Password, OperateAsToUse, ValidateCertificate) + : CredentialsBase.OfCore(epId, ConnectionString, SessionTypeIsPython); return true; } @@ -207,6 +206,8 @@ public string OperateAs { } } + public string OperateAsToUse => _operateAs.Length != 0 ? _operateAs : UserId; + public bool ValidateCertificate { get => _validateCertificate; set { diff --git a/csharp/ExcelAddIn/views/ConnectionManagerDialog.Designer.cs b/csharp/ExcelAddIn/views/ConnectionManagerDialog.Designer.cs index fee2096228f..604c97b2d12 100644 --- a/csharp/ExcelAddIn/views/ConnectionManagerDialog.Designer.cs +++ b/csharp/ExcelAddIn/views/ConnectionManagerDialog.Designer.cs @@ -27,6 +27,10 @@ private void InitializeComponent() { dataGridView1 = new DataGridView(); newButton = new Button(); connectionsLabel = new Label(); + editButton = new Button(); + deleteButton = new Button(); + reconnectButton = new Button(); + makeDefaultButton = new Button(); ((System.ComponentModel.ISupportInitialize)dataGridView1).BeginInit(); SuspendLayout(); // @@ -34,21 +38,25 @@ private void InitializeComponent() { // dataGridView1.AllowUserToAddRows = false; dataGridView1.AllowUserToDeleteRows = false; + dataGridView1.Anchor = AnchorStyles.Top | AnchorStyles.Bottom | AnchorStyles.Left | AnchorStyles.Right; + dataGridView1.AutoSizeColumnsMode = DataGridViewAutoSizeColumnsMode.Fill; dataGridView1.ColumnHeadersHeightSizeMode = DataGridViewColumnHeadersHeightSizeMode.AutoSize; dataGridView1.Location = new Point(68, 83); dataGridView1.Name = "dataGridView1"; dataGridView1.ReadOnly = true; dataGridView1.RowHeadersWidth = 62; + dataGridView1.SelectionMode = DataGridViewSelectionMode.FullRowSelect; dataGridView1.Size = new Size(979, 454); dataGridView1.TabIndex = 0; // // newButton // - newButton.Location = new Point(869, 560); + newButton.Anchor = AnchorStyles.Bottom | AnchorStyles.Right; + newButton.Location = new Point(919, 560); newButton.Name = "newButton"; - newButton.Size = new Size(178, 34); - newButton.TabIndex = 1; - newButton.Text = "New Connection"; + newButton.Size = new Size(128, 34); + newButton.TabIndex = 5; + newButton.Text = "New..."; newButton.UseVisualStyleBackColor = true; newButton.Click += newButton_Click; // @@ -62,11 +70,59 @@ private void InitializeComponent() { connectionsLabel.TabIndex = 2; connectionsLabel.Text = "Connections"; // + // editButton + // + editButton.Anchor = AnchorStyles.Bottom | AnchorStyles.Right; + editButton.Location = new Point(776, 560); + editButton.Name = "editButton"; + editButton.Size = new Size(112, 34); + editButton.TabIndex = 4; + editButton.Text = "Edit..."; + editButton.UseVisualStyleBackColor = true; + editButton.Click += editButton_Click; + // + // deleteButton + // + deleteButton.Anchor = AnchorStyles.Bottom | AnchorStyles.Right; + deleteButton.Location = new Point(339, 560); + deleteButton.Name = "deleteButton"; + deleteButton.Size = new Size(112, 34); + deleteButton.TabIndex = 1; + deleteButton.Text = "Delete"; + deleteButton.UseVisualStyleBackColor = true; + deleteButton.Click += deleteButton_Click; + // + // reconnectButton + // + reconnectButton.Anchor = AnchorStyles.Bottom | AnchorStyles.Right; + reconnectButton.Location = new Point(636, 560); + reconnectButton.Name = "reconnectButton"; + reconnectButton.Size = new Size(112, 34); + reconnectButton.TabIndex = 3; + reconnectButton.Text = "Reconnect"; + reconnectButton.UseVisualStyleBackColor = true; + reconnectButton.Click += reconnectButton_Click; + // + // makeDefaultButton + // + makeDefaultButton.Anchor = AnchorStyles.Bottom | AnchorStyles.Right; + makeDefaultButton.Location = new Point(473, 560); + makeDefaultButton.Name = "makeDefaultButton"; + makeDefaultButton.Size = new Size(139, 34); + makeDefaultButton.TabIndex = 2; + makeDefaultButton.Text = "Make Default"; + makeDefaultButton.UseVisualStyleBackColor = true; + makeDefaultButton.Click += makeDefaultButton_Click; + // // ConnectionManagerDialog // AutoScaleDimensions = new SizeF(10F, 25F); AutoScaleMode = AutoScaleMode.Font; ClientSize = new Size(1115, 615); + Controls.Add(makeDefaultButton); + Controls.Add(reconnectButton); + Controls.Add(deleteButton); + Controls.Add(editButton); Controls.Add(connectionsLabel); Controls.Add(newButton); Controls.Add(dataGridView1); @@ -83,5 +139,9 @@ private void InitializeComponent() { private DataGridView dataGridView1; private Button newButton; private Label connectionsLabel; + private Button editButton; + private Button deleteButton; + private Button reconnectButton; + private Button makeDefaultButton; } } \ No newline at end of file diff --git a/csharp/ExcelAddIn/views/ConnectionManagerDialog.cs b/csharp/ExcelAddIn/views/ConnectionManagerDialog.cs index 750fd39e0de..d56b79f950a 100644 --- a/csharp/ExcelAddIn/views/ConnectionManagerDialog.cs +++ b/csharp/ExcelAddIn/views/ConnectionManagerDialog.cs @@ -2,74 +2,72 @@ namespace Deephaven.ExcelAddIn.Views; +using SelectedRowsAction = Action; + public partial class ConnectionManagerDialog : Form { - private const string IsDefaultColumnName = "IsDefault"; - private const string SettingsButtonColumnName = "settings_button_column"; - private const string ReconnectButtonColumnName = "reconnect_button_column"; - private readonly Action _onNewButtonClicked; - private readonly BindingSource _bindingSource = new(); + public event Action? OnNewButtonClicked; + public event SelectedRowsAction? OnDeleteButtonClicked; + public event SelectedRowsAction? OnReconnectButtonClicked; + public event SelectedRowsAction? OnMakeDefaultButtonClicked; + public event SelectedRowsAction? OnEditButtonClicked; - public ConnectionManagerDialog(Action onNewButtonClicked) { - _onNewButtonClicked = onNewButtonClicked; + private readonly BindingSource _bindingSource = new(); + public ConnectionManagerDialog() { InitializeComponent(); _bindingSource.DataSource = typeof(ConnectionManagerDialogRow); dataGridView1.DataSource = _bindingSource; - - var settingsButtonColumn = new DataGridViewButtonColumn { - Name = SettingsButtonColumnName, - HeaderText = "Credentials", - Text = "Edit", - UseColumnTextForButtonValue = true - }; - - var reconnectButtonColumn = new DataGridViewButtonColumn { - Name = ReconnectButtonColumnName, - HeaderText = "Reconnect", - Text = "Reconnect", - UseColumnTextForButtonValue = true - }; - - dataGridView1.Columns.Add(settingsButtonColumn); - dataGridView1.Columns.Add(reconnectButtonColumn); - - dataGridView1.CellClick += DataGridView1_CellClick; } public void AddRow(ConnectionManagerDialogRow row) { + if (InvokeRequired) { + Invoke(() => AddRow(row)); + return; + } _bindingSource.Add(row); + dataGridView1.ClearSelection(); } - private void DataGridView1_CellClick(object? sender, DataGridViewCellEventArgs e) { - if (e.RowIndex < 0) { + public void RemoveRow(ConnectionManagerDialogRow row) { + if (InvokeRequired) { + Invoke(() => RemoveRow(row)); return; } + _bindingSource.Remove(row); + } - if (_bindingSource[e.RowIndex] is not ConnectionManagerDialogRow row) { - return; - } - var name = dataGridView1.Columns[e.ColumnIndex].Name; + private void newButton_Click(object sender, EventArgs e) { + OnNewButtonClicked?.Invoke(); + } + + private void reconnectButton_Click(object sender, EventArgs e) { + var selections = GetSelectedRows(); + OnReconnectButtonClicked?.Invoke(selections); + } - switch (name) { - case SettingsButtonColumnName: { - row.SettingsClicked(); - break; - } + private void editButton_Click(object sender, EventArgs e) { + var selections = GetSelectedRows(); + OnEditButtonClicked?.Invoke(selections); + } - case ReconnectButtonColumnName: { - row.ReconnectClicked(); - break; - } + private void deleteButton_Click(object sender, EventArgs e) { + var selections = GetSelectedRows(); + OnDeleteButtonClicked?.Invoke(selections); + } - case IsDefaultColumnName: { - row.IsDefaultClicked(); - break; - } - } + private void makeDefaultButton_Click(object sender, EventArgs e) { + var selections = GetSelectedRows(); + OnMakeDefaultButtonClicked?.Invoke(selections); } - private void newButton_Click(object sender, EventArgs e) { - _onNewButtonClicked(); + private ConnectionManagerDialogRow[] GetSelectedRows() { + var result = new List(); + var sr = dataGridView1.SelectedRows; + var count = sr.Count; + for (var i = 0; i != count; ++i) { + result.Add((ConnectionManagerDialogRow)sr[i].DataBoundItem); + } + return result.ToArray(); } } diff --git a/csharp/ExcelAddIn/views/ConnectionManagerDialog.resx b/csharp/ExcelAddIn/views/ConnectionManagerDialog.resx index b3e33e7e100..7f2cf2b8014 100644 --- a/csharp/ExcelAddIn/views/ConnectionManagerDialog.resx +++ b/csharp/ExcelAddIn/views/ConnectionManagerDialog.resx @@ -1,7 +1,7 @@  + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + text/microsoft-resx + + + 2.0 + + + System.Resources.ResXResourceReader, System.Windows.Forms, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089 + + + System.Resources.ResXResourceWriter, System.Windows.Forms, Version=4.0.0.0, Culture=neutral, PublicKeyToken=b77a5c561934e089 + + \ No newline at end of file diff --git a/engine/table/src/main/java/io/deephaven/engine/util/TableDiff.java b/engine/table/src/main/java/io/deephaven/engine/util/TableDiff.java index 179a3b9fc32..156b8219c58 100644 --- a/engine/table/src/main/java/io/deephaven/engine/util/TableDiff.java +++ b/engine/table/src/main/java/io/deephaven/engine/util/TableDiff.java @@ -182,7 +182,7 @@ public enum DiffItems { */ DoublesExact, /** - * Columns that exist in both tables, but in diferent orders are not treated as differences. + * Columns that exist in both tables, but in different orders are not treated as differences. */ ColumnsOrder, /** diff --git a/extensions/iceberg/s3/build.gradle b/extensions/iceberg/s3/build.gradle index c580989fcb2..dfb53c52388 100644 --- a/extensions/iceberg/s3/build.gradle +++ b/extensions/iceberg/s3/build.gradle @@ -26,8 +26,10 @@ dependencies { runtimeOnly libs.awssdk.sts runtimeOnly libs.awssdk.glue - testImplementation libs.junit4 + compileOnly libs.autoservice + annotationProcessor libs.autoservice.compiler + testImplementation libs.junit4 testImplementation project(':engine-test-utils') testImplementation libs.testcontainers diff --git a/extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/IcebergToolsS3.java b/extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/IcebergToolsS3.java index 166b47e5d28..c545ab68540 100644 --- a/extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/IcebergToolsS3.java +++ b/extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/IcebergToolsS3.java @@ -5,11 +5,9 @@ import com.google.common.base.Strings; import org.apache.iceberg.CatalogProperties; -import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.aws.AwsClientProperties; import org.apache.iceberg.aws.glue.GlueCatalog; import org.apache.iceberg.aws.s3.S3FileIOProperties; -import org.apache.iceberg.io.FileIO; import org.apache.iceberg.rest.RESTCatalog; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -68,12 +66,10 @@ public static IcebergCatalogAdapter createS3Rest( properties.put(S3FileIOProperties.ENDPOINT, endpointOverride); } - final FileIO fileIO = CatalogUtil.loadFileIO(S3_FILE_IO_CLASS, properties, null); - final String catalogName = name != null ? name : "IcebergCatalog-" + catalogURI; catalog.initialize(catalogName, properties); - return new IcebergCatalogAdapter(catalog, fileIO); + return new IcebergCatalogAdapter(catalog, properties); } /** @@ -101,11 +97,9 @@ public static IcebergCatalogAdapter createGlue( properties.put(CatalogProperties.URI, catalogURI); properties.put(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation); - final FileIO fileIO = CatalogUtil.loadFileIO(S3_FILE_IO_CLASS, properties, null); - final String catalogName = name != null ? name : "IcebergCatalog-" + catalogURI; catalog.initialize(catalogName, properties); - return new IcebergCatalogAdapter(catalog, fileIO); + return new IcebergCatalogAdapter(catalog, properties); } } diff --git a/extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/S3InstructionsProviderPlugin.java b/extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/S3InstructionsProviderPlugin.java new file mode 100644 index 00000000000..6302ede7ff9 --- /dev/null +++ b/extensions/iceberg/s3/src/main/java/io/deephaven/iceberg/util/S3InstructionsProviderPlugin.java @@ -0,0 +1,54 @@ +// +// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending +// +package io.deephaven.iceberg.util; + +import com.google.auto.service.AutoService; +import io.deephaven.extensions.s3.Credentials; +import io.deephaven.extensions.s3.S3Instructions; +import io.deephaven.iceberg.internal.DataInstructionsProviderPlugin; +import org.apache.iceberg.aws.AwsClientProperties; +import org.apache.iceberg.aws.s3.S3FileIOProperties; +import org.jetbrains.annotations.NotNull; + +import java.net.URI; +import java.util.Map; + +/** + * {@link io.deephaven.iceberg.internal.DataInstructionsProviderPlugin} implementation used for reading files from S3. + */ +@AutoService(io.deephaven.iceberg.internal.DataInstructionsProviderPlugin.class) +@SuppressWarnings("unused") +public final class S3InstructionsProviderPlugin implements DataInstructionsProviderPlugin { + @Override + public Object createInstructions(@NotNull final URI uri, @NotNull final Map properties) { + // If the URI scheme is "s3","s3a","s3n" or if the properties contain one of these specific keys, we can + // create a useful S3Instructions object. + if (uri.getScheme().equals("s3") + || uri.getScheme().equals("s3a") + || uri.getScheme().equals("s3n") + || properties.containsKey(AwsClientProperties.CLIENT_REGION) + || properties.containsKey(S3FileIOProperties.ACCESS_KEY_ID) + || properties.containsKey(S3FileIOProperties.SECRET_ACCESS_KEY) + || properties.containsKey(S3FileIOProperties.ENDPOINT)) { + + final S3Instructions.Builder builder = S3Instructions.builder(); + if (properties.containsKey(AwsClientProperties.CLIENT_REGION)) { + builder.regionName(properties.get(AwsClientProperties.CLIENT_REGION)); + } + if (properties.containsKey(S3FileIOProperties.ENDPOINT)) { + builder.endpointOverride(properties.get(S3FileIOProperties.ENDPOINT)); + } + if (properties.containsKey(S3FileIOProperties.ACCESS_KEY_ID) + && properties.containsKey(S3FileIOProperties.SECRET_ACCESS_KEY)) { + builder.credentials( + Credentials.basic(properties.get(S3FileIOProperties.ACCESS_KEY_ID), + properties.get(S3FileIOProperties.SECRET_ACCESS_KEY))); + } + return builder.build(); + } + + // We have no useful properties for creating an S3Instructions object. + return null; + } +} diff --git a/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergLocalStackTest.java b/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergLocalStackTest.java index de15eceaa04..6683bd42db1 100644 --- a/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergLocalStackTest.java +++ b/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergLocalStackTest.java @@ -8,6 +8,8 @@ import org.junit.BeforeClass; import software.amazon.awssdk.services.s3.S3AsyncClient; +import java.util.Map; + public class IcebergLocalStackTest extends IcebergToolsTest { @BeforeClass @@ -25,4 +27,9 @@ public Builder s3Instructions(final Builder builder) { public S3AsyncClient s3AsyncClient() { return SingletonContainers.LocalStack.s3AsyncClient(); } + + @Override + public Map s3Properties() { + return SingletonContainers.LocalStack.s3Properties(); + } } diff --git a/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergMinIOTest.java b/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergMinIOTest.java index 0e789a64df2..946f3eca90d 100644 --- a/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergMinIOTest.java +++ b/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergMinIOTest.java @@ -10,6 +10,8 @@ import org.junit.BeforeClass; import software.amazon.awssdk.services.s3.S3AsyncClient; +import java.util.Map; + public class IcebergMinIOTest extends IcebergToolsTest { @BeforeClass @@ -29,4 +31,10 @@ public Builder s3Instructions(final Builder builder) { public S3AsyncClient s3AsyncClient() { return SingletonContainers.MinIO.s3AsyncClient(); } + + @Override + public Map s3Properties() { + return SingletonContainers.MinIO.s3Properties(); + } + } diff --git a/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergToolsTest.java b/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergToolsTest.java index 2218e9e3556..eb1640f07c2 100644 --- a/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergToolsTest.java +++ b/extensions/iceberg/s3/src/test/java/io/deephaven/iceberg/util/IcebergToolsTest.java @@ -12,18 +12,16 @@ import io.deephaven.engine.testutil.junit4.EngineCleanup; import io.deephaven.extensions.s3.S3Instructions; import io.deephaven.iceberg.TestCatalog.IcebergTestCatalog; -import io.deephaven.iceberg.TestCatalog.IcebergTestFileIO; import io.deephaven.test.types.OutOfBandTest; import org.apache.iceberg.Snapshot; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.TableIdentifier; -import org.apache.iceberg.io.FileIO; import org.junit.After; import org.junit.Before; import org.junit.Rule; -import org.junit.Test; import org.junit.experimental.categories.Category; +import org.junit.Test; import software.amazon.awssdk.core.async.AsyncRequestBody; import software.amazon.awssdk.services.s3.S3AsyncClient; import software.amazon.awssdk.services.s3.model.CreateBucketRequest; @@ -37,10 +35,7 @@ import java.time.LocalDate; import java.time.LocalDateTime; import java.time.LocalTime; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collection; -import java.util.List; +import java.util.*; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; @@ -106,6 +101,8 @@ public abstract class IcebergToolsTest { public abstract S3Instructions.Builder s3Instructions(S3Instructions.Builder builder); + public abstract Map s3Properties(); + private S3AsyncClient asyncClient; private String bucket; @@ -113,7 +110,6 @@ public abstract class IcebergToolsTest { private String warehousePath; private Catalog resourceCatalog; - private FileIO resourceFileIO; @Rule public final EngineCleanup framework = new EngineCleanup(); @@ -125,10 +121,9 @@ public void setUp() throws ExecutionException, InterruptedException { asyncClient.createBucket(CreateBucketRequest.builder().bucket(bucket).build()).get(); warehousePath = IcebergToolsTest.class.getResource("/warehouse").getPath(); - resourceFileIO = new IcebergTestFileIO("s3://warehouse", warehousePath); // Create the test catalog for the tests - resourceCatalog = IcebergTestCatalog.create(warehousePath, resourceFileIO); + resourceCatalog = IcebergTestCatalog.create(warehousePath, s3Properties()); final S3Instructions s3Instructions = s3Instructions(S3Instructions.builder()).build(); @@ -147,12 +142,12 @@ public void tearDown() throws ExecutionException, InterruptedException { asyncClient.close(); } - private void uploadParquetFiles(final File root, final String prefixToRemove) + private void uploadFiles(final File root, final String prefixToRemove) throws ExecutionException, InterruptedException, TimeoutException { for (final File file : root.listFiles()) { if (file.isDirectory()) { - uploadParquetFiles(file, prefixToRemove); - } else if (file.getName().endsWith(".parquet")) { + uploadFiles(file, prefixToRemove); + } else { final String key = file.getPath().substring(prefixToRemove.length() + 1); keys.add(key); @@ -169,33 +164,33 @@ private void uploadParquetFiles(final File root, final String prefixToRemove) } private void uploadSalesPartitioned() throws ExecutionException, InterruptedException, TimeoutException { - uploadParquetFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sales/sales_partitioned").getPath()), + uploadFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sales/sales_partitioned").getPath()), warehousePath); } private void uploadAllTypes() throws ExecutionException, InterruptedException, TimeoutException { - uploadParquetFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sample/all_types").getPath()), + uploadFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sample/all_types").getPath()), warehousePath); } private void uploadSalesSingle() throws ExecutionException, InterruptedException, TimeoutException { - uploadParquetFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sales/sales_single").getPath()), + uploadFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sales/sales_single").getPath()), warehousePath); } private void uploadSalesMulti() throws ExecutionException, InterruptedException, TimeoutException { - uploadParquetFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sales/sales_multi").getPath()), + uploadFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sales/sales_multi").getPath()), warehousePath); } private void uploadSalesRenamed() throws ExecutionException, InterruptedException, TimeoutException { - uploadParquetFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sales/sales_renamed").getPath()), + uploadFiles(new File(IcebergToolsTest.class.getResource("/warehouse/sales/sales_renamed").getPath()), warehousePath); } @Test public void testListNamespaces() { - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Collection namespaces = adapter.listNamespaces(); final Collection namespaceNames = @@ -212,7 +207,7 @@ public void testListNamespaces() { @Test public void testListTables() { - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); @@ -236,7 +231,7 @@ public void testListTables() { @Test public void testListSnapshots() { - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final TLongArrayList snapshotIds = new TLongArrayList(); final TableIdentifier tableIdentifier = TableIdentifier.of("sales", "sales_multi"); @@ -264,8 +259,7 @@ public void testListSnapshots() { public void testOpenTableA() throws ExecutionException, InterruptedException, TimeoutException { uploadSalesPartitioned(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -287,8 +281,7 @@ public void testOpenTableA() throws ExecutionException, InterruptedException, Ti public void testOpenTableB() throws ExecutionException, InterruptedException, TimeoutException { uploadSalesMulti(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_multi"); @@ -309,8 +302,7 @@ public void testOpenTableB() throws ExecutionException, InterruptedException, Ti public void testOpenTableC() throws ExecutionException, InterruptedException, TimeoutException { uploadSalesSingle(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_single"); @@ -332,7 +324,7 @@ public void testOpenTableC() throws ExecutionException, InterruptedException, Ti public void testOpenTableS3Only() throws ExecutionException, InterruptedException, TimeoutException { uploadSalesPartitioned(); - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -352,8 +344,7 @@ public void testOpenTableDefinition() throws ExecutionException, InterruptedExce .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -380,8 +371,7 @@ public void testOpenTablePartitionTypeException() { .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -424,8 +414,7 @@ public void testOpenTableDefinitionRename() throws ExecutionException, Interrupt .putColumnRenames("month", "__month") .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -454,8 +443,7 @@ public void testSkippedPartitioningColumn() throws ExecutionException, Interrupt .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -484,8 +472,7 @@ public void testReorderedPartitioningColumn() throws ExecutionException, Interru .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -505,8 +492,7 @@ public void testZeroPartitioningColumns() throws ExecutionException, Interrupted .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -533,8 +519,7 @@ public void testIncorrectPartitioningColumns() throws ExecutionException, Interr .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -569,8 +554,7 @@ public void testMissingPartitioningColumns() { .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -598,8 +582,7 @@ public void testOpenTableColumnRename() throws ExecutionException, InterruptedEx .putColumnRenames("Item_Type", "ItemType") .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -617,8 +600,7 @@ public void testOpenTableColumnLegalization() throws ExecutionException, Interru .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_renamed"); @@ -640,8 +622,7 @@ public void testOpenTableColumnLegalizationRename() .putColumnRenames("Units/Sold", "Units_Sold") .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_renamed"); @@ -672,8 +653,7 @@ public void testOpenTableColumnLegalizationPartitionException() { .dataInstructions(instructions.dataInstructions().get()) .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -703,8 +683,7 @@ public void testOpenTableColumnRenamePartitioningColumns() .putColumnRenames("year", "__year") .build(); - final IcebergCatalogAdapter adapter = - IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_partitioned"); @@ -728,7 +707,7 @@ public void testOpenTableColumnRenamePartitioningColumns() public void testOpenTableSnapshot() throws ExecutionException, InterruptedException, TimeoutException { uploadSalesMulti(); - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_multi"); @@ -760,7 +739,7 @@ public void testOpenTableSnapshot() throws ExecutionException, InterruptedExcept public void testOpenTableSnapshotByID() throws ExecutionException, InterruptedException, TimeoutException { uploadSalesMulti(); - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_multi"); @@ -807,7 +786,7 @@ public void testOpenTableSnapshotByID() throws ExecutionException, InterruptedEx public void testOpenAllTypesTable() throws ExecutionException, InterruptedException, TimeoutException { uploadAllTypes(); - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sample"); final TableIdentifier tableId = TableIdentifier.of(ns, "all_types"); @@ -820,7 +799,7 @@ public void testOpenAllTypesTable() throws ExecutionException, InterruptedExcept @Test public void testTableDefinition() { - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_multi"); @@ -845,7 +824,7 @@ public void testTableDefinition() { @Test public void testTableDefinitionTable() { - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); final Namespace ns = Namespace.of("sales"); final TableIdentifier tableId = TableIdentifier.of(ns, "sales_multi"); @@ -878,7 +857,7 @@ public void testTableDefinitionTable() { @Test public void testTableDefinitionWithInstructions() { - final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog, resourceFileIO); + final IcebergCatalogAdapter adapter = IcebergTools.createAdapter(resourceCatalog); IcebergInstructions localInstructions = IcebergInstructions.builder() .dataInstructions(instructions.dataInstructions().get()) diff --git a/extensions/iceberg/src/main/java/io/deephaven/iceberg/internal/DataInstructionsProviderLoader.java b/extensions/iceberg/src/main/java/io/deephaven/iceberg/internal/DataInstructionsProviderLoader.java new file mode 100644 index 00000000000..4ae28e0e044 --- /dev/null +++ b/extensions/iceberg/src/main/java/io/deephaven/iceberg/internal/DataInstructionsProviderLoader.java @@ -0,0 +1,88 @@ +// +// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending +// +package io.deephaven.iceberg.internal; + +import org.jetbrains.annotations.NotNull; + +import java.net.URI; +import java.util.*; + +/** + * A service loader class for loading {@link DataInstructionsProviderPlugin} implementations at runtime which provide + * {@link DataInstructionsProviderLoader} implementations for different URI paths. + */ +public final class DataInstructionsProviderLoader { + /** + * The list of plugins loaded by the {@link ServiceLoader}. + */ + private static volatile List cachedProviders; + + /** + * Ensure that the {@link DataInstructionsProviderPlugin plugins} are loaded exactly once. + */ + private static void ensureProviders() { + if (cachedProviders == null) { + synchronized (DataInstructionsProviderLoader.class) { + if (cachedProviders == null) { + cachedProviders = new ArrayList<>(); + // Load the plugins + for (final DataInstructionsProviderPlugin plugin : ServiceLoader + .load(DataInstructionsProviderPlugin.class)) { + cachedProviders.add(plugin); + } + } + } + } + } + + /** + * Get a {@link DataInstructionsProviderLoader} instance for the given property collection. + * + * @param properties The property collection. + * @return A {@link DataInstructionsProviderLoader} instance. + */ + public static DataInstructionsProviderLoader create(final Map properties) { + ensureProviders(); + return new DataInstructionsProviderLoader(properties); + } + + /** + * The properties collection for this instance. + */ + private final Map properties; + + /** + * The local list of plugins loaded by the {@link ServiceLoader}. + */ + private final List providers; + + /** + * Create a new {@link DataInstructionsProviderLoader} instance for the given property collection. + * + * @param properties The property collection. + */ + private DataInstructionsProviderLoader(final Map properties) { + this.properties = properties; + providers = cachedProviders; + } + + /** + * Create a new data instructions object compatible with reading from and writing to the given URI, using the + * plugins loaded by the {@link ServiceLoader}. For example, for a "S3" URI, we will create an + * {@code S3Instructions} object which can read files from S3. + * + * @param uri The URI + * @return A data instructions object for the given URI or null if one cannot be found + */ + public Object fromServiceLoader(@NotNull final URI uri) { + for (final DataInstructionsProviderPlugin plugin : providers) { + final Object pluginInstructions = plugin.createInstructions(uri, properties); + if (pluginInstructions != null) { + return pluginInstructions; + } + } + // No plugin found for this URI and property collection. + return null; + } +} diff --git a/extensions/iceberg/src/main/java/io/deephaven/iceberg/internal/DataInstructionsProviderPlugin.java b/extensions/iceberg/src/main/java/io/deephaven/iceberg/internal/DataInstructionsProviderPlugin.java new file mode 100644 index 00000000000..b0e48cc9166 --- /dev/null +++ b/extensions/iceberg/src/main/java/io/deephaven/iceberg/internal/DataInstructionsProviderPlugin.java @@ -0,0 +1,20 @@ +// +// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending +// +package io.deephaven.iceberg.internal; + +import org.jetbrains.annotations.NotNull; + +import java.net.URI; +import java.util.Map; + +/** + * A plugin interface for providing {@link DataInstructionsProviderPlugin} implementations for different property + * collections and URI values. Check out {@link DataInstructionsProviderLoader} for more details. + */ +public interface DataInstructionsProviderPlugin { + /** + * Create a data instructions object for the given URI. + */ + Object createInstructions(@NotNull URI uri, @NotNull final Map properties); +} diff --git a/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergBaseLayout.java b/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergBaseLayout.java index f5334cf866c..7bf0f5222a2 100644 --- a/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergBaseLayout.java +++ b/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergBaseLayout.java @@ -11,6 +11,7 @@ import io.deephaven.iceberg.location.IcebergTableParquetLocationKey; import io.deephaven.iceberg.util.IcebergInstructions; import io.deephaven.parquet.table.ParquetInstructions; +import io.deephaven.iceberg.internal.DataInstructionsProviderLoader; import org.apache.iceberg.*; import org.apache.iceberg.io.FileIO; import org.jetbrains.annotations.NotNull; @@ -53,6 +54,11 @@ public abstract class IcebergBaseLayout implements TableLocationKeyFinder cache; + /** + * The data instructions provider for creating instructions from URI and user-supplied properties. + */ + final DataInstructionsProviderLoader dataInstructionsProvider; + /** * The {@link ParquetInstructions} object that will be used to read any Parquet data files in this table. Only * accessed while synchronized on {@code this}. @@ -79,8 +85,16 @@ protected IcebergTableLocationKey locationKey( } } - // Add the data instructions. - instructions.dataInstructions().ifPresent(builder::setSpecialInstructions); + // Add the data instructions if provided as part of the IcebergInstructions. + if (instructions.dataInstructions().isPresent()) { + builder.setSpecialInstructions(instructions.dataInstructions().get()); + } else { + // Attempt to create data instructions from the properties collection and URI. + final Object dataInstructions = dataInstructionsProvider.fromServiceLoader(fileUri); + if (dataInstructions != null) { + builder.setSpecialInstructions(dataInstructions); + } + } parquetInstructions = builder.build(); } @@ -102,12 +116,14 @@ public IcebergBaseLayout( @NotNull final Table table, @NotNull final Snapshot tableSnapshot, @NotNull final FileIO fileIO, - @NotNull final IcebergInstructions instructions) { + @NotNull final IcebergInstructions instructions, + @NotNull final DataInstructionsProviderLoader dataInstructionsProvider) { this.tableDef = tableDef; this.table = table; this.snapshot = tableSnapshot; this.fileIO = fileIO; this.instructions = instructions; + this.dataInstructionsProvider = dataInstructionsProvider; this.cache = new HashMap<>(); } diff --git a/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergFlatLayout.java b/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergFlatLayout.java index ac4c19283f9..fd407d7702e 100644 --- a/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergFlatLayout.java +++ b/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergFlatLayout.java @@ -7,6 +7,7 @@ import io.deephaven.engine.table.impl.locations.impl.TableLocationKeyFinder; import io.deephaven.iceberg.location.IcebergTableLocationKey; import io.deephaven.iceberg.util.IcebergInstructions; +import io.deephaven.iceberg.internal.DataInstructionsProviderLoader; import org.apache.iceberg.*; import org.apache.iceberg.io.FileIO; import org.jetbrains.annotations.NotNull; @@ -30,8 +31,9 @@ public IcebergFlatLayout( @NotNull final Table table, @NotNull final Snapshot tableSnapshot, @NotNull final FileIO fileIO, - @NotNull final IcebergInstructions instructions) { - super(tableDef, table, tableSnapshot, fileIO, instructions); + @NotNull final IcebergInstructions instructions, + @NotNull final DataInstructionsProviderLoader dataInstructionsProvider) { + super(tableDef, table, tableSnapshot, fileIO, instructions, dataInstructionsProvider); } @Override diff --git a/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergKeyValuePartitionedLayout.java b/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergKeyValuePartitionedLayout.java index 47ec05dfd74..f1a3cc9a5ea 100644 --- a/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergKeyValuePartitionedLayout.java +++ b/extensions/iceberg/src/main/java/io/deephaven/iceberg/layout/IcebergKeyValuePartitionedLayout.java @@ -9,6 +9,7 @@ import io.deephaven.engine.table.impl.locations.impl.TableLocationKeyFinder; import io.deephaven.iceberg.location.IcebergTableLocationKey; import io.deephaven.iceberg.util.IcebergInstructions; +import io.deephaven.iceberg.internal.DataInstructionsProviderLoader; import io.deephaven.util.type.TypeUtils; import org.apache.commons.lang3.mutable.MutableInt; import org.apache.iceberg.*; @@ -24,7 +25,7 @@ * a {@link Snapshot} */ public final class IcebergKeyValuePartitionedLayout extends IcebergBaseLayout { - private class ColumnData { + private static class ColumnData { final String name; final Class type; final int index; @@ -52,8 +53,9 @@ public IcebergKeyValuePartitionedLayout( @NotNull final org.apache.iceberg.Snapshot tableSnapshot, @NotNull final FileIO fileIO, @NotNull final PartitionSpec partitionSpec, - @NotNull final IcebergInstructions instructions) { - super(tableDef, table, tableSnapshot, fileIO, instructions); + @NotNull final IcebergInstructions instructions, + @NotNull final DataInstructionsProviderLoader dataInstructionsProvider) { + super(tableDef, table, tableSnapshot, fileIO, instructions, dataInstructionsProvider); // We can assume due to upstream validation that there are no duplicate names (after renaming) that are included // in the output definition, so we can ignore duplicates. diff --git a/extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java b/extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java index 4955ca9223b..b76a750602d 100644 --- a/extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java +++ b/extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergCatalogAdapter.java @@ -18,6 +18,7 @@ import io.deephaven.engine.table.impl.sources.regioned.RegionedTableComponentFactoryImpl; import io.deephaven.engine.updategraph.UpdateSourceRegistrar; import io.deephaven.engine.util.TableTools; +import io.deephaven.iceberg.internal.DataInstructionsProviderLoader; import io.deephaven.iceberg.layout.IcebergFlatLayout; import io.deephaven.iceberg.layout.IcebergKeyValuePartitionedLayout; import io.deephaven.iceberg.location.IcebergTableLocationFactory; @@ -32,7 +33,6 @@ import org.apache.iceberg.catalog.Namespace; import org.apache.iceberg.catalog.SupportsNamespaces; import org.apache.iceberg.catalog.TableIdentifier; -import org.apache.iceberg.io.FileIO; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.jetbrains.annotations.NotNull; @@ -65,16 +65,25 @@ public class IcebergCatalogAdapter { ColumnDefinition.fromGenericType("SnapshotObject", Snapshot.class)); private final Catalog catalog; - private final FileIO fileIO; + + private final DataInstructionsProviderLoader dataInstructionsProvider; + + /** + * Construct an IcebergCatalogAdapter from a catalog. + */ + IcebergCatalogAdapter(@NotNull final Catalog catalog) { + this(catalog, Map.of()); + } /** - * Construct an IcebergCatalogAdapter from a catalog and file IO. + * Construct an IcebergCatalogAdapter from a catalog and property collection. */ IcebergCatalogAdapter( @NotNull final Catalog catalog, - @NotNull final FileIO fileIO) { + @NotNull final Map properties) { this.catalog = catalog; - this.fileIO = fileIO; + + dataInstructionsProvider = DataInstructionsProviderLoader.create(Map.copyOf(properties)); } /** @@ -648,6 +657,54 @@ public Table readTable( return readTable(TableIdentifier.parse(tableIdentifier), instructions); } + /** + * Retrieve a snapshot of an Iceberg table from the Iceberg catalog. + * + * @param tableIdentifier The table identifier to load + * @param tableSnapshotId The snapshot id to load + * @return The loaded table + * @throws IllegalArgumentException if the snapshot with the given id is not found + */ + private Snapshot getTableSnapshot(@NotNull TableIdentifier tableIdentifier, long tableSnapshotId) { + return listSnapshots(tableIdentifier).stream() + .filter(snapshot -> snapshot.snapshotId() == tableSnapshotId) + .findFirst() + .orElseThrow(() -> new IllegalArgumentException( + "Snapshot with id " + tableSnapshotId + " for table " + tableIdentifier + " not found")); + } + + /** + * Read a static snapshot of an Iceberg table from the Iceberg catalog. + * + * @param tableIdentifier The table identifier to load + * @param tableSnapshotId The snapshot id to load + * @return The loaded table + */ + @SuppressWarnings("unused") + public Table readTable(@NotNull final TableIdentifier tableIdentifier, final long tableSnapshotId) { + // Find the snapshot with the given snapshot id + final Snapshot tableSnapshot = getTableSnapshot(tableIdentifier, tableSnapshotId); + + return readTableInternal(tableIdentifier, tableSnapshot, null); + } + + + /** + * Read a static snapshot of an Iceberg table from the Iceberg catalog. + * + * @param tableIdentifier The table identifier to load + * @param tableSnapshotId The snapshot id to load + * @return The loaded table + */ + @SuppressWarnings("unused") + public Table readTable(@NotNull final String tableIdentifier, final long tableSnapshotId) { + final TableIdentifier tableId = TableIdentifier.parse(tableIdentifier); + // Find the snapshot with the given snapshot id + final Snapshot tableSnapshot = getTableSnapshot(tableId, tableSnapshotId); + + return readTableInternal(tableId, tableSnapshot, null); + } + /** * Read a static snapshot of an Iceberg table from the Iceberg catalog. * @@ -738,11 +795,12 @@ private Table readTableInternal( if (partitionSpec.isUnpartitioned()) { // Create the flat layout location key finder - keyFinder = new IcebergFlatLayout(tableDef, table, snapshot, fileIO, userInstructions); + keyFinder = new IcebergFlatLayout(tableDef, table, snapshot, table.io(), userInstructions, + dataInstructionsProvider); } else { // Create the partitioning column location key finder - keyFinder = new IcebergKeyValuePartitionedLayout(tableDef, table, snapshot, fileIO, partitionSpec, - userInstructions); + keyFinder = new IcebergKeyValuePartitionedLayout(tableDef, table, snapshot, table.io(), partitionSpec, + userInstructions, dataInstructionsProvider); } refreshService = null; @@ -772,12 +830,4 @@ private Table readTableInternal( public Catalog catalog() { return catalog; } - - /** - * Returns the underlying Iceberg {@link FileIO} used by this adapter. - */ - @SuppressWarnings("unused") - public FileIO fileIO() { - return fileIO; - } } diff --git a/extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTools.java b/extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTools.java index bcdda326dca..5dd20f699a9 100644 --- a/extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTools.java +++ b/extensions/iceberg/src/main/java/io/deephaven/iceberg/util/IcebergTools.java @@ -3,8 +3,14 @@ // package io.deephaven.iceberg.util; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; import org.apache.iceberg.catalog.Catalog; -import org.apache.iceberg.io.FileIO; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +import java.util.Map; /** * Tools for accessing tables in the Iceberg table format. @@ -12,8 +18,114 @@ public abstract class IcebergTools { @SuppressWarnings("unused") public static IcebergCatalogAdapter createAdapter( - final Catalog catalog, - final FileIO fileIO) { - return new IcebergCatalogAdapter(catalog, fileIO); + final Catalog catalog) { + return new IcebergCatalogAdapter(catalog); + } + + /** + *

+ * Create an Iceberg catalog adapter for an Iceberg catalog created from configuration properties. These properties + * map to the Iceberg catalog Java API properties and are used to create the catalog and file IO implementations. + *

+ *

+ * The minimal set of properties required to create an Iceberg catalog are: + *

    + *
  • {@code "catalog-impl"} or {@code "type"} - the Java catalog implementation to use. When providing + * {@code "catalog-impl"}, the implementing Java class should be provided (e.g. + * {@code "org.apache.iceberg.rest.RESTCatalog"} or {@code "org.apache.iceberg.aws.glue.GlueCatalog")}. Choices for + * {@code "type"} include {@code "hive"}, {@code "hadoop"}, {@code "rest"}, {@code "glue"}, {@code "nessie"}, + * {@code "jdbc"}.
    • + *
  • {@code "uri"} - the URI of the catalog.
    • + *
+ *

+ * Other common properties include: + *

+ *
    + *
  • {@code "warehouse"} - the location of the data warehouse.
    • + *
  • {@code "client.region"} - the region of the AWS client.
    • + *
  • {@code "s3.access-key-id"} - the S3 access key for reading files.
    • + *
  • {@code "s3.secret-access-key"} - the S3 secret access key for reading files.
    • + *
  • {@code "s3.endpoint"} - the S3 endpoint to connect to.
    • + *
+ *

+ * Additional properties for the specific catalog should also be included, such as as S3-specific properties for + * authentication or endpoint overriding. + *

+ * + * @param name the name of the catalog; if omitted, the catalog URI will be used to generate a name + * @param properties a map containing the Iceberg catalog properties to use + * @return the Iceberg catalog adapter + */ + @SuppressWarnings("unused") + public static IcebergCatalogAdapter createAdapter( + @Nullable final String name, + @NotNull final Map properties) { + return createAdapter(name, properties, Map.of()); + } + + /** + *

+ * Create an Iceberg catalog adapter for an Iceberg catalog created from configuration properties. These properties + * map to the Iceberg catalog Java API properties and are used to create the catalog and file IO implementations. + *

+ *

+ * The minimal set of properties required to create an Iceberg catalog are: + *

    + *
  • {@code "catalog-impl"} or {@code "type"} - the Java catalog implementation to use. When providing + * {@code "catalog-impl"}, the implementing Java class should be provided (e.g. + * {@code "org.apache.iceberg.rest.RESTCatalog"} or {@code "org.apache.iceberg.aws.glue.GlueCatalog")}. Choices for + * {@code "type"} include {@code "hive"}, {@code "hadoop"}, {@code "rest"}, {@code "glue"}, {@code "nessie"}, + * {@code "jdbc"}.
    • + *
  • {@code "uri"} - the URI of the catalog.
    • + *
+ *

+ * Other common properties include: + *

+ *
    + *
  • {@code "warehouse"} - the location of the data warehouse.
    • + *
  • {@code "client.region"} - the region of the AWS client.
    • + *
  • {@code "s3.access-key-id"} - the S3 access key for reading files.
    • + *
  • {@code "s3.secret-access-key"} - the S3 secret access key for reading files.
    • + *
  • {@code "s3.endpoint"} - the S3 endpoint to connect to.
    • + *
+ *

+ * Additional properties for the specific catalog should also be included, such as as S3-specific properties for + * authentication or endpoint overriding. + *

+ * + * @param name the name of the catalog; if omitted, the catalog URI will be used to generate a name + * @param properties a map containing the Iceberg catalog properties to use + * @param hadoopConfig a map containing Hadoop configuration properties to use + * @return the Iceberg catalog adapter + */ + @SuppressWarnings("unused") + public static IcebergCatalogAdapter createAdapter( + @Nullable final String name, + @NotNull final Map properties, + @NotNull final Map hadoopConfig) { + // Validate the minimum required properties are set + if (!properties.containsKey(CatalogProperties.CATALOG_IMPL) + && !properties.containsKey(CatalogUtil.ICEBERG_CATALOG_TYPE)) { + throw new IllegalArgumentException( + String.format("Catalog type '%s' or implementation class '%s' is required", + CatalogUtil.ICEBERG_CATALOG_TYPE, CatalogProperties.CATALOG_IMPL)); + } + if (!properties.containsKey(CatalogProperties.URI)) { + throw new IllegalArgumentException(String.format("Catalog URI property '%s' is required", + CatalogProperties.URI)); + } + + final String catalogUri = properties.get(CatalogProperties.URI); + final String catalogName = name != null ? name : "IcebergCatalog-" + catalogUri; + + // Load the Hadoop configuration with the provided properties + final Configuration hadoopConf = new Configuration(); + hadoopConfig.forEach(hadoopConf::set); + + // Create the Iceberg catalog from the properties + final Catalog catalog = CatalogUtil.buildIcebergCatalog(catalogName, properties, hadoopConf); + + return new IcebergCatalogAdapter(catalog, properties); } + } diff --git a/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestCatalog.java b/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestCatalog.java index e62fbd282e0..3d95032e1f8 100644 --- a/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestCatalog.java +++ b/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestCatalog.java @@ -10,7 +10,7 @@ import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; -import org.apache.iceberg.io.FileIO; +import org.jetbrains.annotations.NotNull; import java.io.File; import java.util.*; @@ -19,7 +19,7 @@ public class IcebergTestCatalog implements Catalog, SupportsNamespaces { private final Map> namespaceTableMap; private final Map tableMap; - private IcebergTestCatalog(final String path, final FileIO fileIO) { + private IcebergTestCatalog(final String path, @NotNull final Map properties) { namespaceTableMap = new HashMap<>(); tableMap = new HashMap<>(); @@ -33,7 +33,7 @@ private IcebergTestCatalog(final String path, final FileIO fileIO) { if (tableFile.isDirectory()) { // Second level is table name. final TableIdentifier tableId = TableIdentifier.of(namespace, tableFile.getName()); - final Table table = IcebergTestTable.loadFromMetadata(tableFile.getAbsolutePath(), fileIO); + final Table table = IcebergTestTable.loadFromMetadata(tableFile.getAbsolutePath(), properties); // Add it to the maps. namespaceTableMap.get(namespace).put(tableId, table); @@ -44,8 +44,8 @@ private IcebergTestCatalog(final String path, final FileIO fileIO) { } } - public static IcebergTestCatalog create(final String path, final FileIO fileIO) { - return new IcebergTestCatalog(path, fileIO); + public static IcebergTestCatalog create(final String path, @NotNull final Map properties) { + return new IcebergTestCatalog(path, properties); } @Override diff --git a/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestFileIO.java b/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestFileIO.java deleted file mode 100644 index 03be6ca1b5e..00000000000 --- a/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestFileIO.java +++ /dev/null @@ -1,49 +0,0 @@ -// -// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending -// -package io.deephaven.iceberg.TestCatalog; - -import org.apache.iceberg.inmemory.InMemoryFileIO; -import org.apache.iceberg.io.InputFile; -import org.apache.iceberg.io.OutputFile; - -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.util.HashSet; -import java.util.Set; - -public class IcebergTestFileIO extends InMemoryFileIO { - private final Set inputFiles; - private final String matchPrefix; - private final String replacePrefix; - - public IcebergTestFileIO(final String matchPrefix, final String replacePrefix) { - this.matchPrefix = matchPrefix; - this.replacePrefix = replacePrefix; - inputFiles = new HashSet<>(); - } - - @Override - public InputFile newInputFile(String s) { - if (!inputFiles.contains(s)) { - try { - final String replaced = s.replace(matchPrefix, replacePrefix); - final byte[] data = Files.readAllBytes(Path.of(replaced)); - addFile(s, data); - inputFiles.add(s); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - return super.newInputFile(s); - } - - @Override - public OutputFile newOutputFile(String s) { - return null; - } - - @Override - public void deleteFile(String s) {} -} diff --git a/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestTable.java b/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestTable.java index d1cf5c2ee0e..bcac783def0 100644 --- a/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestTable.java +++ b/extensions/iceberg/src/test/java/io/deephaven/iceberg/TestCatalog/IcebergTestTable.java @@ -3,10 +3,12 @@ // package io.deephaven.iceberg.TestCatalog; +import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.*; import org.apache.iceberg.encryption.EncryptionManager; import org.apache.iceberg.io.FileIO; import org.apache.iceberg.io.LocationProvider; +import org.apache.iceberg.io.ResolvingFileIO; import org.jetbrains.annotations.NotNull; import org.testcontainers.shaded.org.apache.commons.lang3.NotImplementedException; @@ -18,11 +20,14 @@ public class IcebergTestTable implements Table { private final TableMetadata metadata; - private final FileIO fileIO; + private final Map properties; + private final Configuration hadoopConf; + + private IcebergTestTable(@NotNull final String path, @NotNull final Map properties) { + this.properties = properties; + hadoopConf = new Configuration(); - private IcebergTestTable(@NotNull final String path, @NotNull final FileIO fileIO) { final File metadataRoot = new File(path, "metadata"); - this.fileIO = fileIO; final List metadataFiles = new ArrayList<>(); @@ -44,8 +49,10 @@ private IcebergTestTable(@NotNull final String path, @NotNull final FileIO fileI } } - public static IcebergTestTable loadFromMetadata(@NotNull final String path, @NotNull final FileIO fileIO) { - return new IcebergTestTable(path, fileIO); + public static IcebergTestTable loadFromMetadata( + @NotNull final String path, + @NotNull final Map properties) { + return new IcebergTestTable(path, properties); } @Override @@ -214,7 +221,10 @@ public Transaction newTransaction() { @Override public FileIO io() { - return fileIO; + final ResolvingFileIO io = new ResolvingFileIO(); + io.setConf(hadoopConf); + io.initialize(properties); + return io; } @Override diff --git a/extensions/s3/src/test/java/io/deephaven/extensions/s3/testlib/SingletonContainers.java b/extensions/s3/src/test/java/io/deephaven/extensions/s3/testlib/SingletonContainers.java index befd758f980..5d5b550089d 100644 --- a/extensions/s3/src/test/java/io/deephaven/extensions/s3/testlib/SingletonContainers.java +++ b/extensions/s3/src/test/java/io/deephaven/extensions/s3/testlib/SingletonContainers.java @@ -16,6 +16,7 @@ import software.amazon.awssdk.services.s3.S3AsyncClient; import java.net.URI; +import java.util.Map; public final class SingletonContainers { @@ -53,6 +54,14 @@ public static S3AsyncClient s3AsyncClient() { AwsBasicCredentials.create(LOCALSTACK_S3.getAccessKey(), LOCALSTACK_S3.getSecretKey()))) .build(); } + + public static Map s3Properties() { + return Map.of( + "s3.endpoint", LOCALSTACK_S3.getEndpoint().toString(), + "client.region", LOCALSTACK_S3.getRegion(), + "s3.access-key-id", LOCALSTACK_S3.getAccessKey(), + "s3.secret-access-key", LOCALSTACK_S3.getSecretKey()); + } } public static final class MinIO { @@ -87,5 +96,13 @@ public static S3AsyncClient s3AsyncClient() { AwsBasicCredentials.create(MINIO.getUserName(), MINIO.getPassword()))) .build(); } + + public static Map s3Properties() { + return Map.of( + "s3.endpoint", MINIO.getS3URL(), + "client.region", Region.AWS_GLOBAL.toString(), + "s3.access-key-id", MINIO.getUserName(), + "s3.secret-access-key", MINIO.getPassword()); + } } } diff --git a/py/server/deephaven/experimental/iceberg.py b/py/server/deephaven/experimental/iceberg.py index 0a99d3f1880..16fcd08f37d 100644 --- a/py/server/deephaven/experimental/iceberg.py +++ b/py/server/deephaven/experimental/iceberg.py @@ -11,8 +11,11 @@ from deephaven.experimental import s3 from deephaven.table import Table, TableDefinition, TableDefinitionLike +from deephaven.jcompat import j_hashmap + _JIcebergInstructions = jpy.get_type("io.deephaven.iceberg.util.IcebergInstructions") _JIcebergCatalogAdapter = jpy.get_type("io.deephaven.iceberg.util.IcebergCatalogAdapter") +_JIcebergTools = jpy.get_type("io.deephaven.iceberg.util.IcebergTools") # IcebergToolsS3 is an optional library try: @@ -245,3 +248,74 @@ def adapter_aws_glue( except Exception as e: raise DHError(e, "Failed to build Iceberg Catalog Adapter") from e + +def adapter( + name: Optional[str] = None, + properties: Optional[Dict[str, str]] = None, + hadoopConfig: Optional[Dict[str, str]] = None +) -> IcebergCatalogAdapter: + """ + Create an Iceberg catalog adapter from configuration properties. These properties map to the Iceberg catalog Java + API properties and are used to select the catalog and file IO implementations. + + The minimal set of properties required to create an Iceberg catalog are the following: + - `catalog-impl` or `type` - the Java catalog implementation to use. When providing `catalog-impl`, the + implementing Java class should be provided (e.g. `org.apache.iceberg.rest.RESTCatalog` or + `org.apache.iceberg.aws.glue.GlueCatalog`). Choices for `type` include `hive`, `hadoop`, `rest`, `glue`, + `nessie`, `jdbc`. + - `uri` - the URI of the catalog + + Other common properties include: + - `warehouse` - the root path of the data warehouse. + - `client.region` - the region of the AWS client. + - `s3.access-key-id` - the S3 access key for reading files. + - `s3.secret-access-key` - the S3 secret access key for reading files. + - `s3.endpoint` - the S3 endpoint to connect to. + + Example usage #1 - REST catalog with an S3 backend (using MinIO): + ``` + from deephaven.experimental import iceberg + + adapter = iceberg.adapter_generic(name="generic-adapter", properties={ + "type" : "rest", + "uri" : "http://rest:8181", + "client.region" : "us-east-1", + "s3.access-key-id" : "admin", + "s3.secret-access-key" : "password", + "s3.endpoint" : "http://minio:9000" + }) + ``` + + Example usage #2 - AWS Glue catalog: + ``` + from deephaven.experimental import iceberg + + ## Note: region and credential information are loaded by the catalog from the environment + adapter = iceberg.adapter_generic(name="generic-adapter", properties={ + "type" : "glue", + "uri" : "s3://lab-warehouse/sales", + }); + ``` + + Args: + name (Optional[str]): a descriptive name of the catalog; if omitted the catalog name is inferred from the + catalog URI property. + properties (Optional[Dict[str, str]]): the properties of the catalog to load + hadoopConfig (Optional[Dict[str, str]]): hadoop configuration properties for the catalog to load + + Returns: + IcebergCatalogAdapter: the catalog adapter created from the provided properties + + Raises: + DHError: If unable to build the catalog adapter + """ + + try: + return IcebergCatalogAdapter( + _JIcebergTools.createAdapter( + name, + j_hashmap(properties if properties is not None else {}), + j_hashmap(hadoopConfig if hadoopConfig is not None else {}))) + except Exception as e: + raise DHError(e, "Failed to build Iceberg Catalog Adapter") from e +