[GitHub] [accumulo] cshannon commented on a diff in pull request #3250: Added On-Demand table feature

["cshannon (via GitHub)" <gi...@apache.org>]

cshannon commented on code in PR #3250:
URL: https://github.com/apache/accumulo/pull/3250#discussion_r1148421248


##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -1285,4 +1304,153 @@ public BlockCacheConfiguration getBlockCacheConfiguration(AccumuloConfiguration
     return BlockCacheConfiguration.forTabletServer(acuConf);
   }
 
+  public int getOnDemandOnlineUnloadedForLowMemory() {
+    return onDemandUnloadedLowMemory.get();
+  }
+
+  // called from AssignmentHandler
+  public void insertOnDemandAccessTime(KeyExtent extent) {
+    onDemandTabletAccessTimes.putIfAbsent(extent, new AtomicLong(System.nanoTime()));
+  }
+
+  // called from getOnlineExtent
+  private void updateOnDemandAccessTime(KeyExtent extent) {
+    final long currentTime = System.nanoTime();
+    AtomicLong l = onDemandTabletAccessTimes.get(extent);
+    if (l != null) {
+      l.set(currentTime);
+    }
+  }
+
+  // called from UnloadTabletHandler
+  public void removeOnDemandAccessTime(KeyExtent extent) {
+    onDemandTabletAccessTimes.remove(extent);
+  }
+
+  private boolean isTabletInUse(KeyExtent extent) {
+    // Don't call getOnlineTablet as that will update the last access time
+    final Tablet t = onlineTablets.snapshot().get(extent);
+    if (t == null) {
+      return false;
+    }
+    t.updateRates(System.currentTimeMillis());
+    if (t.ingestRate() != 0.0 && t.queryRate() != 0.0 && t.scanRate() != 0.0) {
+      // tablet is ingesting or scanning
+      return true;
+    }
+    return false;
+  }
+
+  public void evaluateOnDemandTabletsForUnload() {
+
+    final SortedMap<KeyExtent,Tablet> online = getOnlineTablets();
+
+    // Find and remove access time entries for KeyExtents
+    // that are no longer in the onlineTablets collection
+    Set<KeyExtent> missing = onDemandTabletAccessTimes.keySet().stream()
+        .filter(k -> !online.containsKey(k)).collect(Collectors.toSet());
+    if (!missing.isEmpty()) {
+      log.debug("Removing onDemandAccessTimes for tablets as tablets no longer online: {}",
+          missing);
+      missing.forEach(onDemandTabletAccessTimes::remove);
+      if (onDemandTabletAccessTimes.isEmpty()) {
+        return;
+      }
+    }
+
+    // It's possible, from a tablet split or merge for example,
+    // that there is an onDemand tablet that is hosted for which
+    // we have no access time. Add any missing online onDemand
+    // tablets
+    online.forEach((k, v) -> {
+      if (v.isOnDemand() && !onDemandTabletAccessTimes.containsKey(k)) {
+        insertOnDemandAccessTime(k);
+      }
+    });
+
+    log.debug("Evaluating online onDemand tablets: {}", onDemandTabletAccessTimes);
+
+    if (onDemandTabletAccessTimes.isEmpty()) {
+      return;
+    }
+
+    // If the TabletServer is running low on memory, don't call the SPI
+    // plugin to evaluate which onDemand tablets to unload, just get the
+    // onDemand tablet with the oldest access time and unload it.
+    if (getContext().getLowMemoryDetector().isRunningLowOnMemory()) {
+      final SortedMap<Long,KeyExtent> timeSortedOnDemandExtents = new TreeMap<>();
+      onDemandTabletAccessTimes.forEach((k, v) -> timeSortedOnDemandExtents.put(v.get(), k));
+      Long oldestAccessTime = timeSortedOnDemandExtents.firstKey();
+      KeyExtent oldestKeyExtent = timeSortedOnDemandExtents.get(oldestAccessTime);
+      log.warn("Unloading onDemand tablet: {} for table: {} due to low memory", oldestKeyExtent,
+          oldestKeyExtent.tableId());
+      getContext().getAmple().mutateTablet(oldestKeyExtent).deleteOnDemand().mutate();
+      onDemandUnloadedLowMemory.addAndGet(1);
+      return;
+    }
+
+    // onDemandTabletAccessTimes is a HashMap. Sort the extents
+    // so that we can process them by table.
+    final SortedMap<KeyExtent,AtomicLong> sortedOnDemandExtents =
+        new TreeMap<KeyExtent,AtomicLong>();
+    sortedOnDemandExtents.putAll(onDemandTabletAccessTimes);
+
+    // The access times are updated when getOnlineTablet is called by other methods,
+    // but may not necessarily capture whether or not the Tablet is currently being used.
+    // For example, getOnlineTablet is called from startScan but not from continueScan.
+    // Instead of instrumenting all of the locations where the tablet is touched we
+    // can use the Tablet metrics.
+    final Set<KeyExtent> onDemandTabletsInUse = new HashSet<>();
+    for (KeyExtent extent : sortedOnDemandExtents.keySet()) {
+      if (isTabletInUse(extent)) {
+        onDemandTabletsInUse.add(extent);
+      }
+    }
+    if (!onDemandTabletsInUse.isEmpty()) {
+      log.debug("Removing onDemandAccessTimes for tablets as tablets are in use: {}",
+          onDemandTabletsInUse);
+      onDemandTabletsInUse.forEach(sortedOnDemandExtents::remove);
+      if (sortedOnDemandExtents.isEmpty()) {
+        return;
+      }
+    }
+
+    Set<TableId> tableIds = sortedOnDemandExtents.keySet().stream().map((k) -> {
+      return k.tableId();
+    }).distinct().collect(Collectors.toSet());
+    log.debug("Tables that have online onDemand tablets: {}", tableIds);
+    final Map<TableId,OnDemandTabletUnloader> unloaders = new HashMap<>();
+    tableIds.forEach(tid -> {
+      TableConfiguration tconf = getContext().getTableConfiguration(tid);
+      String tableContext = ClassLoaderUtil.tableContext(tconf);
+      String unloaderClassName = tconf.get(Property.TABLE_ONDEMAND_UNLOADER);
+      try {
+        Class<? extends OnDemandTabletUnloader> clazz = ClassLoaderUtil.loadClass(tableContext,
+            unloaderClassName, OnDemandTabletUnloader.class);
+        unloaders.put(tid, clazz.getConstructor().newInstance());
+      } catch (ClassNotFoundException | InstantiationException | IllegalAccessException
+          | IllegalArgumentException | InvocationTargetException | NoSuchMethodException
+          | SecurityException e) {
+        log.error(
+            "Error constructing OnDemandTabletUnloader implementation, not unloading onDemand tablets",
+            e);
+        return;
+      }
+    });
+    tableIds.forEach(tid -> {
+      Map<KeyExtent,AtomicLong> subset = sortedOnDemandExtents.entrySet().stream().filter((e) -> {
+        return e.getKey().tableId().equals(tid);
+      }).collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue()));

Review Comment:
   ```suggestion
             AtomicLong> subset = sortedOnDemandExtents.entrySet().stream()
                 .filter((e) -> e.getKey().tableId().equals(tid))
                 .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
   ```



##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java:
##########
@@ -1555,4 +1556,21 @@ public TSummaries contiuneGetSummaries(TInfo tinfo, long sessionId)
       return handleTimeout(sessionId);
     }
   }
+
+  @Override
+  public void bringOnDemandTabletsOnline(TInfo tinfo, TCredentials credentials, String tableId,
+      List<TKeyExtent> extents) throws ThriftSecurityException, TException {
+    final TableId tid = TableId.of(tableId);
+    NamespaceId namespaceId = getNamespaceId(credentials, tid);
+    if (!security.canScan(credentials, tid, namespaceId)) {
+      throw new ThriftSecurityException(credentials.getPrincipal(),
+          SecurityErrorCode.PERMISSION_DENIED);
+    }
+    try (TabletsMutator mutator = this.context.getAmple().mutateTablets()) {
+      extents.forEach(e -> {
+        mutator.mutateTablet(KeyExtent.fromThrift(e)).putOnDemand().mutate();
+      });

Review Comment:
   ```suggestion
         extents.forEach(e -> mutator.mutateTablet(KeyExtent.fromThrift(e)).putOnDemand().mutate());
   ```



##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java:
##########
@@ -524,6 +541,72 @@ public TabletLocation locateTablet(ClientContext context, Text row, boolean skip
     }
   }
 
+  @Override
+  public long onDemandTabletsOnlined() {
+    return onDemandTabletsOnlinedCount.get();
+  }
+
+  private void bringOnDemandTabletsOnline(ClientContext context, Range range)
+      throws AccumuloException, AccumuloSecurityException {
+
+    // Confirm that table is in an onDemand state. Don't throw an exception
+    // if the table is not found, calling code will already handle it.
+    try {
+      String tableName = context.getTableName(tableId);
+      if (!context.tableOperations().isOnDemand(tableName)) {
+        log.trace("bringOnDemandTabletsOnline: table {} is not in ondemand state", tableId);
+        return;
+      }
+    } catch (TableNotFoundException e) {
+      log.trace("bringOnDemandTabletsOnline: table not found: {}", tableId);
+      return;
+    }
+
+    final Text scanRangeStart = range.getStartKey().getRow();
+    final Text scanRangeEnd = range.getEndKey().getRow();
+    // Turn the scan range into a KeyExtent and bring online all ondemand tablets
+    // that are overlapped by the scan range
+    final KeyExtent scanRangeKE = new KeyExtent(tableId, scanRangeEnd, scanRangeStart);
+
+    List<TKeyExtent> extentsToBringOnline = new ArrayList<>();
+
+    TabletsMetadata m = context.getAmple().readTablets().forTable(tableId)
+        .overlapping(scanRangeStart, true, null).build();
+    for (TabletMetadata tm : m) {
+      final KeyExtent tabletExtent = tm.getExtent();
+      log.trace("Evaluating tablet {} against range {}", tabletExtent, scanRangeKE);
+      if (tm.getEndRow() != null && tm.getEndRow().compareTo(scanRangeStart) < 0) {
+        // the end row of this tablet is before the start row, skip it
+        log.trace("tablet {} is before scan start range: {}", tabletExtent, scanRangeStart);
+        continue;
+      }
+      if (tm.getPrevEndRow() != null && tm.getPrevEndRow().compareTo(scanRangeEnd) > 0) {
+        // the start row of this tablet is after the scan range end row, skip it
+        log.trace("tablet {} is after scan end range: {}", tabletExtent, scanRangeEnd);
+        continue;
+      }
+      if (scanRangeKE.overlaps(tabletExtent)) {
+        if (!tm.getOnDemand()) {
+          Location loc = tm.getLocation();
+          if (loc != null) {
+            log.debug("tablet {} has location of: {}:{}", loc.getType(), loc.getHostPort());

Review Comment:
   ```suggestion
               log.debug("tablet {} has location of: {}:{}", tabletExtent, loc.getType(),
                   loc.getHostPort());
   ```



##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletClientHandler.java:
##########
@@ -1555,4 +1556,21 @@ public TSummaries contiuneGetSummaries(TInfo tinfo, long sessionId)
       return handleTimeout(sessionId);
     }
   }
+
+  @Override
+  public void bringOnDemandTabletsOnline(TInfo tinfo, TCredentials credentials, String tableId,
+      List<TKeyExtent> extents) throws ThriftSecurityException, TException {

Review Comment:
   ```suggestion
         List<TKeyExtent> extents) throws TException {
   ```



##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.server.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.manager.state.tables.TableState;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.metadata.TabletLocationState;
+import org.apache.accumulo.core.metadata.TabletState;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+  private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+  public static void main(String[] args) throws Exception {
+    ServerUtilOpts opts = new ServerUtilOpts();
+    opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+    Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+    try (Scope scope = span.makeCurrent()) {
+      ServerContext context = opts.getServerContext();
+      findOnline(context, null);
+    } finally {
+      span.end();
+    }
+  }
+
+  static int findOnline(ServerContext context, String tableName) throws TableNotFoundException {
+
+    final AtomicBoolean scanning = new AtomicBoolean(false);
+
+    LiveTServerSet tservers = new LiveTServerSet(context, new Listener() {
+      @Override
+      public void update(LiveTServerSet current, Set<TServerInstance> deleted,
+          Set<TServerInstance> added) {
+        if (!deleted.isEmpty() && scanning.get()) {
+          log.warn("Tablet servers deleted while scanning: {}", deleted);
+        }
+        if (!added.isEmpty() && scanning.get()) {
+          log.warn("Tablet servers added while scanning: {}", added);
+        }
+      }
+    });
+    tservers.startListeningForTabletServerChanges();
+    scanning.set(true);
+
+    System.out.println("Scanning " + MetadataTable.NAME);
+
+    Range range = TabletsSection.getRange();
+    if (tableName != null) {
+      TableId tableId = context.getTableId(tableName);
+      range = new KeyExtent(tableId, null, null).toMetaRange();
+    }
+
+    try (MetaDataTableScanner metaScanner =
+        new MetaDataTableScanner(context, range, MetadataTable.NAME)) {
+      return checkTablets(context, metaScanner, tservers);
+    }
+  }
+
+  private static int checkTablets(ServerContext context, Iterator<TabletLocationState> scanner,
+      LiveTServerSet tservers) {
+    int online = 0;
+
+    while (scanner.hasNext() && !System.out.checkError()) {
+      TabletLocationState locationState = scanner.next();
+      TabletState state = locationState.getState(tservers.getCurrentServers());
+      if (state != null && state == TabletState.HOSTED
+          && context.getTableManager().getTableState(locationState.extent.tableId())

Review Comment:
   ```suggestion
         if (state == TabletState.HOSTED
             && context.getTableManager().getTableState(locationState.extent.tableId())
   ```



##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.server.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.manager.state.tables.TableState;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.metadata.TabletLocationState;
+import org.apache.accumulo.core.metadata.TabletState;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+  private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+  public static void main(String[] args) throws Exception {
+    ServerUtilOpts opts = new ServerUtilOpts();
+    opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+    Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+    try (Scope scope = span.makeCurrent()) {
+      ServerContext context = opts.getServerContext();
+      findOnline(context, null);

Review Comment:
   I assume the intent is to allow optionally passing in a table name as an argument? Right now this is always null and nowhere else is the method called.



##########
server/base/src/main/java/org/apache/accumulo/server/util/ListOnlineOnDemandTablets.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.accumulo.server.util;
+
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.TableId;
+import org.apache.accumulo.core.dataImpl.KeyExtent;
+import org.apache.accumulo.core.manager.state.tables.TableState;
+import org.apache.accumulo.core.metadata.MetadataTable;
+import org.apache.accumulo.core.metadata.TServerInstance;
+import org.apache.accumulo.core.metadata.TabletLocationState;
+import org.apache.accumulo.core.metadata.TabletState;
+import org.apache.accumulo.core.metadata.schema.MetadataSchema.TabletsSection;
+import org.apache.accumulo.core.trace.TraceUtil;
+import org.apache.accumulo.server.ServerContext;
+import org.apache.accumulo.server.cli.ServerUtilOpts;
+import org.apache.accumulo.server.manager.LiveTServerSet;
+import org.apache.accumulo.server.manager.LiveTServerSet.Listener;
+import org.apache.accumulo.server.manager.state.MetaDataTableScanner;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
+public class ListOnlineOnDemandTablets {
+  private static final Logger log = LoggerFactory.getLogger(ListOnlineOnDemandTablets.class);
+
+  public static void main(String[] args) throws Exception {
+    ServerUtilOpts opts = new ServerUtilOpts();
+    opts.parseArgs(ListOnlineOnDemandTablets.class.getName(), args);
+    Span span = TraceUtil.startSpan(ListOnlineOnDemandTablets.class, "main");
+    try (Scope scope = span.makeCurrent()) {
+      ServerContext context = opts.getServerContext();
+      findOnline(context, null);
+    } finally {
+      span.end();
+    }
+  }
+
+  static int findOnline(ServerContext context, String tableName) throws TableNotFoundException {
+
+    final AtomicBoolean scanning = new AtomicBoolean(false);
+
+    LiveTServerSet tservers = new LiveTServerSet(context, new Listener() {
+      @Override
+      public void update(LiveTServerSet current, Set<TServerInstance> deleted,
+          Set<TServerInstance> added) {
+        if (!deleted.isEmpty() && scanning.get()) {
+          log.warn("Tablet servers deleted while scanning: {}", deleted);
+        }
+        if (!added.isEmpty() && scanning.get()) {
+          log.warn("Tablet servers added while scanning: {}", added);
+        }
+      }
+    });
+    tservers.startListeningForTabletServerChanges();
+    scanning.set(true);
+
+    System.out.println("Scanning " + MetadataTable.NAME);
+
+    Range range = TabletsSection.getRange();
+    if (tableName != null) {
+      TableId tableId = context.getTableId(tableName);
+      range = new KeyExtent(tableId, null, null).toMetaRange();
+    }
+
+    try (MetaDataTableScanner metaScanner =
+        new MetaDataTableScanner(context, range, MetadataTable.NAME)) {
+      return checkTablets(context, metaScanner, tservers);
+    }
+  }
+
+  private static int checkTablets(ServerContext context, Iterator<TabletLocationState> scanner,
+      LiveTServerSet tservers) {
+    int online = 0;
+
+    while (scanner.hasNext() && !System.out.checkError()) {
+      TabletLocationState locationState = scanner.next();
+      TabletState state = locationState.getState(tservers.getCurrentServers());
+      if (state != null && state == TabletState.HOSTED
+          && context.getTableManager().getTableState(locationState.extent.tableId())

Review Comment:
   Small nit: Null check is redundant here 



##########
server/tserver/src/main/java/org/apache/accumulo/tserver/TabletServer.java:
##########
@@ -1285,4 +1304,153 @@ public BlockCacheConfiguration getBlockCacheConfiguration(AccumuloConfiguration
     return BlockCacheConfiguration.forTabletServer(acuConf);
   }
 
+  public int getOnDemandOnlineUnloadedForLowMemory() {
+    return onDemandUnloadedLowMemory.get();
+  }
+
+  // called from AssignmentHandler
+  public void insertOnDemandAccessTime(KeyExtent extent) {
+    onDemandTabletAccessTimes.putIfAbsent(extent, new AtomicLong(System.nanoTime()));
+  }
+
+  // called from getOnlineExtent
+  private void updateOnDemandAccessTime(KeyExtent extent) {
+    final long currentTime = System.nanoTime();
+    AtomicLong l = onDemandTabletAccessTimes.get(extent);
+    if (l != null) {
+      l.set(currentTime);
+    }
+  }
+
+  // called from UnloadTabletHandler
+  public void removeOnDemandAccessTime(KeyExtent extent) {
+    onDemandTabletAccessTimes.remove(extent);
+  }
+
+  private boolean isTabletInUse(KeyExtent extent) {
+    // Don't call getOnlineTablet as that will update the last access time
+    final Tablet t = onlineTablets.snapshot().get(extent);
+    if (t == null) {
+      return false;
+    }
+    t.updateRates(System.currentTimeMillis());
+    if (t.ingestRate() != 0.0 && t.queryRate() != 0.0 && t.scanRate() != 0.0) {
+      // tablet is ingesting or scanning
+      return true;
+    }
+    return false;
+  }
+
+  public void evaluateOnDemandTabletsForUnload() {
+
+    final SortedMap<KeyExtent,Tablet> online = getOnlineTablets();
+
+    // Find and remove access time entries for KeyExtents
+    // that are no longer in the onlineTablets collection
+    Set<KeyExtent> missing = onDemandTabletAccessTimes.keySet().stream()
+        .filter(k -> !online.containsKey(k)).collect(Collectors.toSet());
+    if (!missing.isEmpty()) {
+      log.debug("Removing onDemandAccessTimes for tablets as tablets no longer online: {}",
+          missing);
+      missing.forEach(onDemandTabletAccessTimes::remove);
+      if (onDemandTabletAccessTimes.isEmpty()) {
+        return;
+      }
+    }
+
+    // It's possible, from a tablet split or merge for example,
+    // that there is an onDemand tablet that is hosted for which
+    // we have no access time. Add any missing online onDemand
+    // tablets
+    online.forEach((k, v) -> {
+      if (v.isOnDemand() && !onDemandTabletAccessTimes.containsKey(k)) {
+        insertOnDemandAccessTime(k);
+      }
+    });
+
+    log.debug("Evaluating online onDemand tablets: {}", onDemandTabletAccessTimes);
+
+    if (onDemandTabletAccessTimes.isEmpty()) {
+      return;
+    }
+
+    // If the TabletServer is running low on memory, don't call the SPI
+    // plugin to evaluate which onDemand tablets to unload, just get the
+    // onDemand tablet with the oldest access time and unload it.
+    if (getContext().getLowMemoryDetector().isRunningLowOnMemory()) {
+      final SortedMap<Long,KeyExtent> timeSortedOnDemandExtents = new TreeMap<>();
+      onDemandTabletAccessTimes.forEach((k, v) -> timeSortedOnDemandExtents.put(v.get(), k));
+      Long oldestAccessTime = timeSortedOnDemandExtents.firstKey();
+      KeyExtent oldestKeyExtent = timeSortedOnDemandExtents.get(oldestAccessTime);
+      log.warn("Unloading onDemand tablet: {} for table: {} due to low memory", oldestKeyExtent,
+          oldestKeyExtent.tableId());
+      getContext().getAmple().mutateTablet(oldestKeyExtent).deleteOnDemand().mutate();
+      onDemandUnloadedLowMemory.addAndGet(1);
+      return;
+    }
+
+    // onDemandTabletAccessTimes is a HashMap. Sort the extents
+    // so that we can process them by table.
+    final SortedMap<KeyExtent,AtomicLong> sortedOnDemandExtents =
+        new TreeMap<KeyExtent,AtomicLong>();
+    sortedOnDemandExtents.putAll(onDemandTabletAccessTimes);
+
+    // The access times are updated when getOnlineTablet is called by other methods,
+    // but may not necessarily capture whether or not the Tablet is currently being used.
+    // For example, getOnlineTablet is called from startScan but not from continueScan.
+    // Instead of instrumenting all of the locations where the tablet is touched we
+    // can use the Tablet metrics.
+    final Set<KeyExtent> onDemandTabletsInUse = new HashSet<>();
+    for (KeyExtent extent : sortedOnDemandExtents.keySet()) {
+      if (isTabletInUse(extent)) {
+        onDemandTabletsInUse.add(extent);
+      }
+    }
+    if (!onDemandTabletsInUse.isEmpty()) {
+      log.debug("Removing onDemandAccessTimes for tablets as tablets are in use: {}",
+          onDemandTabletsInUse);
+      onDemandTabletsInUse.forEach(sortedOnDemandExtents::remove);
+      if (sortedOnDemandExtents.isEmpty()) {
+        return;
+      }
+    }
+
+    Set<TableId> tableIds = sortedOnDemandExtents.keySet().stream().map((k) -> {
+      return k.tableId();
+    }).distinct().collect(Collectors.toSet());
+    log.debug("Tables that have online onDemand tablets: {}", tableIds);
+    final Map<TableId,OnDemandTabletUnloader> unloaders = new HashMap<>();
+    tableIds.forEach(tid -> {
+      TableConfiguration tconf = getContext().getTableConfiguration(tid);
+      String tableContext = ClassLoaderUtil.tableContext(tconf);
+      String unloaderClassName = tconf.get(Property.TABLE_ONDEMAND_UNLOADER);
+      try {
+        Class<? extends OnDemandTabletUnloader> clazz = ClassLoaderUtil.loadClass(tableContext,
+            unloaderClassName, OnDemandTabletUnloader.class);
+        unloaders.put(tid, clazz.getConstructor().newInstance());
+      } catch (ClassNotFoundException | InstantiationException | IllegalAccessException
+          | IllegalArgumentException | InvocationTargetException | NoSuchMethodException
+          | SecurityException e) {
+        log.error(
+            "Error constructing OnDemandTabletUnloader implementation, not unloading onDemand tablets",
+            e);
+        return;
+      }
+    });
+    tableIds.forEach(tid -> {
+      Map<KeyExtent,AtomicLong> subset = sortedOnDemandExtents.entrySet().stream().filter((e) -> {
+        return e.getKey().tableId().equals(tid);
+      }).collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue()));

Review Comment:
   This is just stylistic so don't need to change it if you don't want



##########
core/src/main/java/org/apache/accumulo/core/clientImpl/TabletLocatorImpl.java:
##########
@@ -524,6 +541,72 @@ public TabletLocation locateTablet(ClientContext context, Text row, boolean skip
     }
   }
 
+  @Override
+  public long onDemandTabletsOnlined() {
+    return onDemandTabletsOnlinedCount.get();
+  }
+
+  private void bringOnDemandTabletsOnline(ClientContext context, Range range)
+      throws AccumuloException, AccumuloSecurityException {
+
+    // Confirm that table is in an onDemand state. Don't throw an exception
+    // if the table is not found, calling code will already handle it.
+    try {
+      String tableName = context.getTableName(tableId);
+      if (!context.tableOperations().isOnDemand(tableName)) {
+        log.trace("bringOnDemandTabletsOnline: table {} is not in ondemand state", tableId);
+        return;
+      }
+    } catch (TableNotFoundException e) {
+      log.trace("bringOnDemandTabletsOnline: table not found: {}", tableId);
+      return;
+    }
+
+    final Text scanRangeStart = range.getStartKey().getRow();
+    final Text scanRangeEnd = range.getEndKey().getRow();
+    // Turn the scan range into a KeyExtent and bring online all ondemand tablets
+    // that are overlapped by the scan range
+    final KeyExtent scanRangeKE = new KeyExtent(tableId, scanRangeEnd, scanRangeStart);
+
+    List<TKeyExtent> extentsToBringOnline = new ArrayList<>();
+
+    TabletsMetadata m = context.getAmple().readTablets().forTable(tableId)
+        .overlapping(scanRangeStart, true, null).build();
+    for (TabletMetadata tm : m) {
+      final KeyExtent tabletExtent = tm.getExtent();
+      log.trace("Evaluating tablet {} against range {}", tabletExtent, scanRangeKE);
+      if (tm.getEndRow() != null && tm.getEndRow().compareTo(scanRangeStart) < 0) {
+        // the end row of this tablet is before the start row, skip it
+        log.trace("tablet {} is before scan start range: {}", tabletExtent, scanRangeStart);
+        continue;
+      }
+      if (tm.getPrevEndRow() != null && tm.getPrevEndRow().compareTo(scanRangeEnd) > 0) {
+        // the start row of this tablet is after the scan range end row, skip it
+        log.trace("tablet {} is after scan end range: {}", tabletExtent, scanRangeEnd);
+        continue;
+      }
+      if (scanRangeKE.overlaps(tabletExtent)) {
+        if (!tm.getOnDemand()) {
+          Location loc = tm.getLocation();
+          if (loc != null) {
+            log.debug("tablet {} has location of: {}:{}", loc.getType(), loc.getHostPort());

Review Comment:
   There is a mismatched number of args here for the log statement, the tabletExtent is missing



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: notifications-unsubscribe@accumulo.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org