asterixdb-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From amo...@apache.org
Subject [10/12] asterixdb git commit: Cleanup FileSplit and FileReference
Date Thu, 24 Nov 2016 06:30:14 GMT
http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadata.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadata.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadata.java
index 04006f1..3a486a5 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadata.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadata.java
@@ -18,24 +18,26 @@
  */
 package org.apache.asterix.transaction.management.resource;
 
-import java.io.File;
 import java.util.Map;
 
 import org.apache.asterix.common.context.BaseOperationTracker;
 import org.apache.asterix.common.ioopcallbacks.LSMBTreeWithBuddyIOOperationCallbackFactory;
 import org.apache.asterix.common.transactions.IAsterixAppRuntimeContextProvider;
+import org.apache.asterix.common.transactions.Resource;
 import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IIOManager;
 import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeUtils;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
+import org.apache.hyracks.storage.common.file.LocalResource;
 
 /**
  * The local resource for disk only lsm btree with buddy tree
  */
-public class ExternalBTreeWithBuddyLocalResourceMetadata extends AbstractLSMLocalResourceMetadata {
+public class ExternalBTreeWithBuddyLocalResourceMetadata extends Resource {
 
     private static final long serialVersionUID = 1L;
 
@@ -45,10 +47,11 @@ public class ExternalBTreeWithBuddyLocalResourceMetadata extends AbstractLSMLoca
     private final Map<String, String> mergePolicyProperties;
     private final int[] buddyBtreeFields;
 
-    public ExternalBTreeWithBuddyLocalResourceMetadata(int datasetID, IBinaryComparatorFactory[] btreeCmpFactories,
+    public ExternalBTreeWithBuddyLocalResourceMetadata(int datasetID, int partition,
+            IBinaryComparatorFactory[] btreeCmpFactories,
             ITypeTraits[] typeTraits, ILSMMergePolicyFactory mergePolicyFactory,
             Map<String, String> mergePolicyProperties, int[] buddyBtreeFields) {
-        super(datasetID, null, null, null);
+        super(datasetID, partition, null, null, null);
         this.btreeCmpFactories = btreeCmpFactories;
         this.typeTraits = typeTraits;
         this.mergePolicyFactory = mergePolicyFactory;
@@ -57,16 +60,17 @@ public class ExternalBTreeWithBuddyLocalResourceMetadata extends AbstractLSMLoca
     }
 
     @Override
-    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider, String filePath,
-                                         int partition, int ioDeviceNum) throws HyracksDataException {
-        FileReference file = new FileReference(new File(filePath));
-        return LSMBTreeUtils.createExternalBTreeWithBuddy(file, runtimeContextProvider.getBufferCache(),
+    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider,
+            LocalResource resource) throws HyracksDataException {
+        IIOManager ioManager = runtimeContextProvider.getIOManager();
+        FileReference file = ioManager.getFileRef(resource.getPath(), true);
+        return LSMBTreeUtils.createExternalBTreeWithBuddy(ioManager, file, runtimeContextProvider.getBufferCache(),
                 runtimeContextProvider.getFileMapManager(), typeTraits, btreeCmpFactories,
                 runtimeContextProvider.getBloomFilterFalsePositiveRate(),
                 mergePolicyFactory.createMergePolicy(mergePolicyProperties,
                         runtimeContextProvider.getDatasetLifecycleManager()),
-                new BaseOperationTracker(datasetID,
-                        runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetID)),
+                new BaseOperationTracker(datasetId(),
+                        runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetId())),
                 runtimeContextProvider.getLSMIOScheduler(),
                 LSMBTreeWithBuddyIOOperationCallbackFactory.INSTANCE.createIOOperationCallback(), buddyBtreeFields, -1,
                 true);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadataFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadataFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadataFactory.java
new file mode 100644
index 0000000..836bcf4
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalBTreeWithBuddyLocalResourceMetadataFactory.java
@@ -0,0 +1,54 @@
+/*
+ * 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
+ *
+ *   http://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.asterix.transaction.management.resource;
+
+import java.util.Map;
+
+import org.apache.asterix.common.transactions.Resource;
+import org.apache.asterix.common.transactions.ResourceFactory;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
+
+public class ExternalBTreeWithBuddyLocalResourceMetadataFactory extends ResourceFactory {
+    private static final long serialVersionUID = 1L;
+    private final ITypeTraits[] typeTraits;
+    private final IBinaryComparatorFactory[] btreeCmpFactories;
+    private final ILSMMergePolicyFactory mergePolicyFactory;
+    private final Map<String, String> mergePolicyProperties;
+    private final int[] buddyBtreeFields;
+
+    public ExternalBTreeWithBuddyLocalResourceMetadataFactory(int datasetID,
+            IBinaryComparatorFactory[] btreeCmpFactories,
+            ITypeTraits[] typeTraits, ILSMMergePolicyFactory mergePolicyFactory,
+            Map<String, String> mergePolicyProperties, int[] buddyBtreeFields) {
+        super(datasetID, null, null, null);
+        this.btreeCmpFactories = btreeCmpFactories;
+        this.typeTraits = typeTraits;
+        this.mergePolicyFactory = mergePolicyFactory;
+        this.mergePolicyProperties = mergePolicyProperties;
+        this.buddyBtreeFields = buddyBtreeFields;
+    }
+
+    @Override
+    public Resource resource(int partition) {
+        return new ExternalBTreeWithBuddyLocalResourceMetadata(datasetId, partition, btreeCmpFactories, typeTraits,
+                mergePolicyFactory, mergePolicyProperties, buddyBtreeFields);
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadata.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadata.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadata.java
index 9ed15d2..7f6c4a3 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadata.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadata.java
@@ -18,7 +18,6 @@
  */
 package org.apache.asterix.transaction.management.resource;
 
-import java.io.File;
 import java.util.Map;
 
 import org.apache.asterix.common.context.BaseOperationTracker;
@@ -29,12 +28,14 @@ import org.apache.hyracks.api.dataflow.value.ILinearizeComparatorFactory;
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IIOManager;
 import org.apache.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory;
 import org.apache.hyracks.storage.am.common.api.TreeIndexException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
 import org.apache.hyracks.storage.am.lsm.rtree.utils.LSMRTreeUtils;
 import org.apache.hyracks.storage.am.rtree.frames.RTreePolicyType;
+import org.apache.hyracks.storage.common.file.LocalResource;
 
 /**
  * The local resource class for disk only lsm r-tree
@@ -46,25 +47,26 @@ public class ExternalRTreeLocalResourceMetadata extends LSMRTreeLocalResourceMet
     public ExternalRTreeLocalResourceMetadata(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] rtreeCmpFactories,
             IBinaryComparatorFactory[] btreeCmpFactories, IPrimitiveValueProviderFactory[] valueProviderFactories,
             RTreePolicyType rtreePolicyType, ILinearizeComparatorFactory linearizeCmpFactory, int datasetID,
-            ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties, int[] btreeFields,
-            boolean isPointMBR) {
+            int partition, ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
+            int[] btreeFields, boolean isPointMBR) {
         super(typeTraits, rtreeCmpFactories, btreeCmpFactories, valueProviderFactories, rtreePolicyType,
-                linearizeCmpFactory, datasetID, mergePolicyFactory, mergePolicyProperties, null, null, null,
+                linearizeCmpFactory, datasetID, partition, mergePolicyFactory, mergePolicyProperties, null, null, null,
                 btreeFields, null, isPointMBR);
     }
 
     @Override
-    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider, String filePath,
-            int partition, int ioDeviceNum) throws HyracksDataException {
-        FileReference file = new FileReference(new File(filePath));
+    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider,
+            LocalResource resource) throws HyracksDataException {
+        IIOManager ioManager = runtimeContextProvider.getIOManager();
+        FileReference file = ioManager.getFileRef(resource.getPath(), true);
         try {
-            return LSMRTreeUtils.createExternalRTree(file, runtimeContextProvider.getBufferCache(),
+            return LSMRTreeUtils.createExternalRTree(ioManager, file, runtimeContextProvider.getBufferCache(),
                     runtimeContextProvider.getFileMapManager(), typeTraits, rtreeCmpFactories, btreeCmpFactories,
                     valueProviderFactories, rtreePolicyType, runtimeContextProvider.getBloomFilterFalsePositiveRate(),
                     mergePolicyFactory.createMergePolicy(mergePolicyProperties,
                             runtimeContextProvider.getDatasetLifecycleManager()),
-                    new BaseOperationTracker(datasetID,
-                            runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetID)),
+                    new BaseOperationTracker(datasetId(),
+                            runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetId())),
                     runtimeContextProvider.getLSMIOScheduler(),
                     LSMRTreeIOOperationCallbackFactory.INSTANCE.createIOOperationCallback(), linearizeCmpFactory,
                     btreeFields, -1, true, isPointMBR);
@@ -72,5 +74,4 @@ public class ExternalRTreeLocalResourceMetadata extends LSMRTreeLocalResourceMet
             throw new HyracksDataException(e);
         }
     }
-
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadataFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadataFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadataFactory.java
new file mode 100644
index 0000000..076fef1
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/ExternalRTreeLocalResourceMetadataFactory.java
@@ -0,0 +1,53 @@
+/*
+ * 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
+ *
+ *   http://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.asterix.transaction.management.resource;
+
+import java.util.Map;
+
+import org.apache.asterix.common.transactions.Resource;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ILinearizeComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
+import org.apache.hyracks.storage.am.rtree.frames.RTreePolicyType;
+
+public class ExternalRTreeLocalResourceMetadataFactory extends LSMRTreeLocalResourceMetadataFactory {
+
+    private static final long serialVersionUID = 1L;
+
+    public ExternalRTreeLocalResourceMetadataFactory(ITypeTraits[] typeTraits,
+            IBinaryComparatorFactory[] rtreeCmpFactories,
+            IBinaryComparatorFactory[] btreeCmpFactories, IPrimitiveValueProviderFactory[] valueProviderFactories,
+            RTreePolicyType rtreePolicyType, ILinearizeComparatorFactory linearizeCmpFactory, int datasetID,
+            ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
+            int[] btreeFields, boolean isPointMBR) {
+        super(typeTraits, rtreeCmpFactories, btreeCmpFactories, valueProviderFactories, rtreePolicyType,
+                linearizeCmpFactory, datasetID, mergePolicyFactory, mergePolicyProperties, null, null, null,
+                btreeFields, null, isPointMBR);
+    }
+
+    @Override
+    public Resource resource(int partition) {
+        return new ExternalRTreeLocalResourceMetadata(filterTypeTraits, rtreeCmpFactories, btreeCmpFactories,
+                valueProviderFactories, rtreePolicyType, linearizeCmpFactory, datasetId, partition, mergePolicyFactory,
+                mergePolicyProperties, btreeFields, isPointMBR);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java
index b1db1f3..9c52241 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadata.java
@@ -18,22 +18,27 @@
  */
 package org.apache.asterix.transaction.management.resource;
 
-import java.io.File;
+import java.util.List;
 import java.util.Map;
 
 import org.apache.asterix.common.api.IDatasetLifecycleManager;
 import org.apache.asterix.common.context.BaseOperationTracker;
 import org.apache.asterix.common.ioopcallbacks.LSMBTreeIOOperationCallbackFactory;
 import org.apache.asterix.common.transactions.IAsterixAppRuntimeContextProvider;
+import org.apache.asterix.common.transactions.Resource;
 import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IIOManager;
+import org.apache.hyracks.api.io.IODeviceHandle;
 import org.apache.hyracks.storage.am.lsm.btree.impls.LSMBTree;
 import org.apache.hyracks.storage.am.lsm.btree.util.LSMBTreeUtils;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
+import org.apache.hyracks.storage.common.file.LocalResource;
 
-public class LSMBTreeLocalResourceMetadata extends AbstractLSMLocalResourceMetadata {
+public class LSMBTreeLocalResourceMetadata extends Resource {
 
     private static final long serialVersionUID = 1L;
 
@@ -46,10 +51,11 @@ public class LSMBTreeLocalResourceMetadata extends AbstractLSMLocalResourceMetad
     protected final int[] btreeFields;
 
     public LSMBTreeLocalResourceMetadata(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] cmpFactories,
-            int[] bloomFilterKeyFields, boolean isPrimary, int datasetID, ILSMMergePolicyFactory mergePolicyFactory,
+            int[] bloomFilterKeyFields, boolean isPrimary, int datasetID, int partition,
+            ILSMMergePolicyFactory mergePolicyFactory,
             Map<String, String> mergePolicyProperties, ITypeTraits[] filterTypeTraits,
             IBinaryComparatorFactory[] filterCmpFactories, int[] btreeFields, int[] filterFields) {
-        super(datasetID, filterTypeTraits, filterCmpFactories, filterFields);
+        super(datasetID, partition, filterTypeTraits, filterCmpFactories, filterFields);
         this.typeTraits = typeTraits;
         this.cmpFactories = cmpFactories;
         this.bloomFilterKeyFields = bloomFilterKeyFields;
@@ -60,20 +66,38 @@ public class LSMBTreeLocalResourceMetadata extends AbstractLSMLocalResourceMetad
     }
 
     @Override
-    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider, String filePath,
-                                         int partition, int ioDeviceNum) {
-        FileReference file = new FileReference(new File(filePath));
+    public String toString() {
+        return new StringBuilder().append(" { \"").append(LSMBTreeLocalResourceMetadata.class.getName()).append(
+                "\" : {").append("\"datasetId\" : ").append(datasetId()).append(", \"partition\" : ").append(
+                        partition()).append(" } ").append(" }").toString();
+    }
+
+    @Override
+    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider,
+            LocalResource resource) throws HyracksDataException {
+        IIOManager ioManager = runtimeContextProvider.getIOManager();
+        FileReference file = ioManager.getFileRef(resource.getPath(), true);
+        List<IODeviceHandle> ioDevices = ioManager.getIODevices();
+        int ioDeviceNum = 0;
+        for (int i = 0; i < ioDevices.size(); i++) {
+            IODeviceHandle device = ioDevices.get(i);
+            if (device == file.getDeviceHandle()) {
+                ioDeviceNum = i;
+                break;
+            }
+        }
         final IDatasetLifecycleManager datasetLifecycleManager = runtimeContextProvider.getDatasetLifecycleManager();
-        LSMBTree lsmBTree = LSMBTreeUtils.createLSMTree(datasetLifecycleManager.getVirtualBufferCaches(datasetID,
+        LSMBTree lsmBTree = LSMBTreeUtils.createLSMTree(ioManager, datasetLifecycleManager.getVirtualBufferCaches(
+                datasetId(),
                 ioDeviceNum), file, runtimeContextProvider.getBufferCache(), runtimeContextProvider.getFileMapManager(),
-                typeTraits, cmpFactories, bloomFilterKeyFields, runtimeContextProvider.getBloomFilterFalsePositiveRate(),
+                typeTraits, cmpFactories, bloomFilterKeyFields, runtimeContextProvider
+                        .getBloomFilterFalsePositiveRate(),
                 mergePolicyFactory.createMergePolicy(mergePolicyProperties, datasetLifecycleManager),
-                isPrimary ? runtimeContextProvider.getLSMBTreeOperationTracker(datasetID)
-                        : new BaseOperationTracker(datasetID, datasetLifecycleManager.getDatasetInfo(datasetID)),
+                isPrimary ? runtimeContextProvider.getLSMBTreeOperationTracker(datasetId())
+                        : new BaseOperationTracker(datasetId(), datasetLifecycleManager.getDatasetInfo(datasetId())),
                 runtimeContextProvider.getLSMIOScheduler(),
                 LSMBTreeIOOperationCallbackFactory.INSTANCE.createIOOperationCallback(), isPrimary, filterTypeTraits,
                 filterCmpFactories, btreeFields, filterFields, true);
         return lsmBTree;
     }
-
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadataFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadataFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadataFactory.java
new file mode 100644
index 0000000..34d646d
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMBTreeLocalResourceMetadataFactory.java
@@ -0,0 +1,61 @@
+/*
+ * 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
+ *
+ *   http://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.asterix.transaction.management.resource;
+
+import java.util.Map;
+
+import org.apache.asterix.common.transactions.Resource;
+import org.apache.asterix.common.transactions.ResourceFactory;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
+
+public class LSMBTreeLocalResourceMetadataFactory extends ResourceFactory {
+
+    private static final long serialVersionUID = 1L;
+    protected final ITypeTraits[] typeTraits;
+    protected final IBinaryComparatorFactory[] cmpFactories;
+    protected final int[] bloomFilterKeyFields;
+    protected final boolean isPrimary;
+    protected final ILSMMergePolicyFactory mergePolicyFactory;
+    protected final Map<String, String> mergePolicyProperties;
+    protected final int[] btreeFields;
+
+    public LSMBTreeLocalResourceMetadataFactory(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] cmpFactories,
+            int[] bloomFilterKeyFields, boolean isPrimary, int datasetID,
+            ILSMMergePolicyFactory mergePolicyFactory,
+            Map<String, String> mergePolicyProperties, ITypeTraits[] filterTypeTraits,
+            IBinaryComparatorFactory[] filterCmpFactories, int[] btreeFields, int[] filterFields) {
+        super(datasetID, filterTypeTraits, filterCmpFactories, filterFields);
+        this.typeTraits = typeTraits;
+        this.cmpFactories = cmpFactories;
+        this.bloomFilterKeyFields = bloomFilterKeyFields;
+        this.isPrimary = isPrimary;
+        this.mergePolicyFactory = mergePolicyFactory;
+        this.mergePolicyProperties = mergePolicyProperties;
+        this.btreeFields = btreeFields;
+    }
+
+    @Override
+    public Resource resource(int partition) {
+        return new LSMBTreeLocalResourceMetadata(typeTraits, cmpFactories, bloomFilterKeyFields, isPrimary, datasetId,
+                partition, mergePolicyFactory, mergePolicyProperties, typeTraits, filterCmpFactories, btreeFields,
+                filterFields);
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java
index 87319b4..06bf298 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadata.java
@@ -24,17 +24,22 @@ import java.util.Map;
 import org.apache.asterix.common.context.BaseOperationTracker;
 import org.apache.asterix.common.ioopcallbacks.LSMInvertedIndexIOOperationCallbackFactory;
 import org.apache.asterix.common.transactions.IAsterixAppRuntimeContextProvider;
+import org.apache.asterix.common.transactions.Resource;
 import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IIOManager;
+import org.apache.hyracks.api.io.IODeviceHandle;
 import org.apache.hyracks.storage.am.common.api.IndexException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
 import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
 import org.apache.hyracks.storage.am.lsm.invertedindex.tokenizers.IBinaryTokenizerFactory;
 import org.apache.hyracks.storage.am.lsm.invertedindex.util.InvertedIndexUtils;
+import org.apache.hyracks.storage.common.file.LocalResource;
 
-public class LSMInvertedIndexLocalResourceMetadata extends AbstractLSMLocalResourceMetadata {
+public class LSMInvertedIndexLocalResourceMetadata extends Resource {
 
     private static final long serialVersionUID = 1L;
 
@@ -53,11 +58,11 @@ public class LSMInvertedIndexLocalResourceMetadata extends AbstractLSMLocalResou
     public LSMInvertedIndexLocalResourceMetadata(ITypeTraits[] invListTypeTraits,
             IBinaryComparatorFactory[] invListCmpFactories, ITypeTraits[] tokenTypeTraits,
             IBinaryComparatorFactory[] tokenCmpFactories, IBinaryTokenizerFactory tokenizerFactory,
-            boolean isPartitioned, int datasetID, ILSMMergePolicyFactory mergePolicyFactory,
+            boolean isPartitioned, int datasetID, int partition, ILSMMergePolicyFactory mergePolicyFactory,
             Map<String, String> mergePolicyProperties, ITypeTraits[] filterTypeTraits,
             IBinaryComparatorFactory[] filterCmpFactories, int[] invertedIndexFields, int[] filterFields,
             int[] filterFieldsForNonBulkLoadOps, int[] invertedIndexFieldsForNonBulkLoadOps) {
-        super(datasetID, filterTypeTraits, filterCmpFactories, filterFields);
+        super(datasetID, partition, filterTypeTraits, filterCmpFactories, filterFields);
         this.invListTypeTraits = invListTypeTraits;
         this.invListCmpFactories = invListCmpFactories;
         this.tokenTypeTraits = tokenTypeTraits;
@@ -72,33 +77,44 @@ public class LSMInvertedIndexLocalResourceMetadata extends AbstractLSMLocalResou
     }
 
     @Override
-    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider, String filePath,
-                                         int partition, int ioDeviceNum) throws HyracksDataException {
+    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider,
+            LocalResource resource) throws HyracksDataException {
+        IIOManager ioManager = runtimeContextProvider.getIOManager();
+        FileReference file = ioManager.getFileRef(resource.getPath(), true);
+        List<IODeviceHandle> ioDevices = ioManager.getIODevices();
+        int ioDeviceNum = 0;
+        for (int i = 0; i < ioDevices.size(); i++) {
+            IODeviceHandle device = ioDevices.get(i);
+            if (device == file.getDeviceHandle()) {
+                ioDeviceNum = i;
+                break;
+            }
+        }
         List<IVirtualBufferCache> virtualBufferCaches = runtimeContextProvider.getDatasetLifecycleManager()
-                .getVirtualBufferCaches(datasetID, ioDeviceNum);
+                .getVirtualBufferCaches(datasetId(), ioDeviceNum);
         try {
             if (isPartitioned) {
-                return InvertedIndexUtils.createPartitionedLSMInvertedIndex(virtualBufferCaches,
+                return InvertedIndexUtils.createPartitionedLSMInvertedIndex(ioManager, virtualBufferCaches,
                         runtimeContextProvider.getFileMapManager(), invListTypeTraits, invListCmpFactories,
                         tokenTypeTraits, tokenCmpFactories, tokenizerFactory, runtimeContextProvider.getBufferCache(),
-                        filePath, runtimeContextProvider.getBloomFilterFalsePositiveRate(),
+                        file.getAbsolutePath(), runtimeContextProvider.getBloomFilterFalsePositiveRate(),
                         mergePolicyFactory.createMergePolicy(mergePolicyProperties,
                                 runtimeContextProvider.getDatasetLifecycleManager()),
-                        new BaseOperationTracker(datasetID,
-                                runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetID)),
+                        new BaseOperationTracker(datasetId(),
+                                runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetId())),
                         runtimeContextProvider.getLSMIOScheduler(),
                         LSMInvertedIndexIOOperationCallbackFactory.INSTANCE.createIOOperationCallback(),
                         invertedIndexFields, filterTypeTraits, filterCmpFactories, filterFields,
                         filterFieldsForNonBulkLoadOps, invertedIndexFieldsForNonBulkLoadOps, true);
             } else {
-                return InvertedIndexUtils.createLSMInvertedIndex(virtualBufferCaches,
+                return InvertedIndexUtils.createLSMInvertedIndex(ioManager, virtualBufferCaches,
                         runtimeContextProvider.getFileMapManager(), invListTypeTraits, invListCmpFactories,
                         tokenTypeTraits, tokenCmpFactories, tokenizerFactory, runtimeContextProvider.getBufferCache(),
-                        filePath, runtimeContextProvider.getBloomFilterFalsePositiveRate(),
+                        file.getAbsolutePath(), runtimeContextProvider.getBloomFilterFalsePositiveRate(),
                         mergePolicyFactory.createMergePolicy(mergePolicyProperties,
                                 runtimeContextProvider.getDatasetLifecycleManager()),
-                        new BaseOperationTracker(datasetID,
-                                runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetID)),
+                        new BaseOperationTracker(datasetId(),
+                                runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetId())),
                         runtimeContextProvider.getLSMIOScheduler(),
                         LSMInvertedIndexIOOperationCallbackFactory.INSTANCE.createIOOperationCallback(),
                         invertedIndexFields, filterTypeTraits, filterCmpFactories, filterFields,

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadataFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadataFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadataFactory.java
new file mode 100644
index 0000000..a5a6c2a
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMInvertedIndexLocalResourceMetadataFactory.java
@@ -0,0 +1,74 @@
+/*
+ * 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
+ *
+ *   http://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.asterix.transaction.management.resource;
+
+import java.util.Map;
+
+import org.apache.asterix.common.transactions.Resource;
+import org.apache.asterix.common.transactions.ResourceFactory;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
+import org.apache.hyracks.storage.am.lsm.invertedindex.tokenizers.IBinaryTokenizerFactory;
+
+public class LSMInvertedIndexLocalResourceMetadataFactory extends ResourceFactory {
+
+    private static final long serialVersionUID = 1L;
+    private final ITypeTraits[] invListTypeTraits;
+    private final IBinaryComparatorFactory[] invListCmpFactories;
+    private final ITypeTraits[] tokenTypeTraits;
+    private final IBinaryComparatorFactory[] tokenCmpFactories;
+    private final IBinaryTokenizerFactory tokenizerFactory;
+    private final boolean isPartitioned;
+    private final ILSMMergePolicyFactory mergePolicyFactory;
+    private final Map<String, String> mergePolicyProperties;
+    private final int[] invertedIndexFields;
+    private final int[] filterFieldsForNonBulkLoadOps;
+    private final int[] invertedIndexFieldsForNonBulkLoadOps;
+
+    public LSMInvertedIndexLocalResourceMetadataFactory(ITypeTraits[] invListTypeTraits,
+            IBinaryComparatorFactory[] invListCmpFactories, ITypeTraits[] tokenTypeTraits,
+            IBinaryComparatorFactory[] tokenCmpFactories, IBinaryTokenizerFactory tokenizerFactory,
+            boolean isPartitioned, int datasetID, ILSMMergePolicyFactory mergePolicyFactory,
+            Map<String, String> mergePolicyProperties, ITypeTraits[] filterTypeTraits,
+            IBinaryComparatorFactory[] filterCmpFactories, int[] invertedIndexFields, int[] filterFields,
+            int[] filterFieldsForNonBulkLoadOps, int[] invertedIndexFieldsForNonBulkLoadOps) {
+        super(datasetID, filterTypeTraits, filterCmpFactories, filterFields);
+        this.invListTypeTraits = invListTypeTraits;
+        this.invListCmpFactories = invListCmpFactories;
+        this.tokenTypeTraits = tokenTypeTraits;
+        this.tokenCmpFactories = tokenCmpFactories;
+        this.tokenizerFactory = tokenizerFactory;
+        this.isPartitioned = isPartitioned;
+        this.mergePolicyFactory = mergePolicyFactory;
+        this.mergePolicyProperties = mergePolicyProperties;
+        this.invertedIndexFields = invertedIndexFields;
+        this.filterFieldsForNonBulkLoadOps = filterFieldsForNonBulkLoadOps;
+        this.invertedIndexFieldsForNonBulkLoadOps = invertedIndexFieldsForNonBulkLoadOps;
+    }
+
+    @Override
+    public Resource resource(int partition) {
+        return new LSMInvertedIndexLocalResourceMetadata(invListTypeTraits, invListCmpFactories, tokenTypeTraits,
+                tokenCmpFactories, tokenizerFactory, isPartitioned, datasetId, partition, mergePolicyFactory,
+                mergePolicyProperties, filterTypeTraits, filterCmpFactories, invertedIndexFields,
+                filterFields, filterFieldsForNonBulkLoadOps, invertedIndexFieldsForNonBulkLoadOps);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java
index 535448c..73c2dd9 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadata.java
@@ -18,18 +18,20 @@
  */
 package org.apache.asterix.transaction.management.resource;
 
-import java.io.File;
 import java.util.List;
 import java.util.Map;
 
 import org.apache.asterix.common.context.BaseOperationTracker;
 import org.apache.asterix.common.ioopcallbacks.LSMRTreeIOOperationCallbackFactory;
 import org.apache.asterix.common.transactions.IAsterixAppRuntimeContextProvider;
+import org.apache.asterix.common.transactions.Resource;
 import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
 import org.apache.hyracks.api.dataflow.value.ILinearizeComparatorFactory;
 import org.apache.hyracks.api.dataflow.value.ITypeTraits;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
 import org.apache.hyracks.api.io.FileReference;
+import org.apache.hyracks.api.io.IIOManager;
+import org.apache.hyracks.api.io.IODeviceHandle;
 import org.apache.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory;
 import org.apache.hyracks.storage.am.common.api.TreeIndexException;
 import org.apache.hyracks.storage.am.lsm.common.api.ILSMIndex;
@@ -37,8 +39,9 @@ import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
 import org.apache.hyracks.storage.am.lsm.common.api.IVirtualBufferCache;
 import org.apache.hyracks.storage.am.lsm.rtree.utils.LSMRTreeUtils;
 import org.apache.hyracks.storage.am.rtree.frames.RTreePolicyType;
+import org.apache.hyracks.storage.common.file.LocalResource;
 
-public class LSMRTreeLocalResourceMetadata extends AbstractLSMLocalResourceMetadata {
+public class LSMRTreeLocalResourceMetadata extends Resource {
 
     private static final long serialVersionUID = 1L;
 
@@ -57,10 +60,10 @@ public class LSMRTreeLocalResourceMetadata extends AbstractLSMLocalResourceMetad
     public LSMRTreeLocalResourceMetadata(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] rtreeCmpFactories,
             IBinaryComparatorFactory[] btreeCmpFactories, IPrimitiveValueProviderFactory[] valueProviderFactories,
             RTreePolicyType rtreePolicyType, ILinearizeComparatorFactory linearizeCmpFactory, int datasetID,
-            ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
+            int partition, ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
             ITypeTraits[] filterTypeTraits, IBinaryComparatorFactory[] filterCmpFactories, int[] rtreeFields,
             int[] btreeFields, int[] filterFields, boolean isPointMBR) {
-        super(datasetID, filterTypeTraits, filterCmpFactories, filterFields);
+        super(datasetID, partition, filterTypeTraits, filterCmpFactories, filterFields);
         this.typeTraits = typeTraits;
         this.rtreeCmpFactories = rtreeCmpFactories;
         this.btreeCmpFactories = btreeCmpFactories;
@@ -75,19 +78,29 @@ public class LSMRTreeLocalResourceMetadata extends AbstractLSMLocalResourceMetad
     }
 
     @Override
-    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider, String filePath,
-            int partition, int ioDeviceNum) throws HyracksDataException {
-        FileReference file = new FileReference(new File(filePath));
+    public ILSMIndex createIndexInstance(IAsterixAppRuntimeContextProvider runtimeContextProvider,
+            LocalResource resource) throws HyracksDataException {
+        IIOManager ioManager = runtimeContextProvider.getIOManager();
+        FileReference file = ioManager.getFileRef(resource.getPath(), true);
+        List<IODeviceHandle> ioDevices = ioManager.getIODevices();
+        int ioDeviceNum = 0;
+        for (int i = 0; i < ioDevices.size(); i++) {
+            IODeviceHandle device = ioDevices.get(i);
+            if (device == file.getDeviceHandle()) {
+                ioDeviceNum = i;
+                break;
+            }
+        }
         List<IVirtualBufferCache> virtualBufferCaches = runtimeContextProvider.getDatasetLifecycleManager()
-                .getVirtualBufferCaches(datasetID, ioDeviceNum);
+                .getVirtualBufferCaches(datasetId(), ioDeviceNum);
         try {
-            return LSMRTreeUtils.createLSMTreeWithAntiMatterTuples(virtualBufferCaches, file,
+            return LSMRTreeUtils.createLSMTreeWithAntiMatterTuples(ioManager, virtualBufferCaches, file,
                     runtimeContextProvider.getBufferCache(), runtimeContextProvider.getFileMapManager(), typeTraits,
                     rtreeCmpFactories, btreeCmpFactories, valueProviderFactories, rtreePolicyType,
                     mergePolicyFactory.createMergePolicy(mergePolicyProperties,
                             runtimeContextProvider.getDatasetLifecycleManager()),
-                    new BaseOperationTracker(datasetID,
-                            runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetID)),
+                    new BaseOperationTracker(datasetId(),
+                            runtimeContextProvider.getDatasetLifecycleManager().getDatasetInfo(datasetId())),
                     runtimeContextProvider.getLSMIOScheduler(),
                     LSMRTreeIOOperationCallbackFactory.INSTANCE.createIOOperationCallback(), linearizeCmpFactory,
                     rtreeFields, filterTypeTraits, filterCmpFactories, filterFields, true, isPointMBR);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadataFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadataFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadataFactory.java
new file mode 100644
index 0000000..0ecf0aa
--- /dev/null
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/LSMRTreeLocalResourceMetadataFactory.java
@@ -0,0 +1,75 @@
+/*
+ * 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
+ *
+ *   http://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.asterix.transaction.management.resource;
+
+import java.util.Map;
+
+import org.apache.asterix.common.transactions.Resource;
+import org.apache.asterix.common.transactions.ResourceFactory;
+import org.apache.hyracks.api.dataflow.value.IBinaryComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ILinearizeComparatorFactory;
+import org.apache.hyracks.api.dataflow.value.ITypeTraits;
+import org.apache.hyracks.storage.am.common.api.IPrimitiveValueProviderFactory;
+import org.apache.hyracks.storage.am.lsm.common.api.ILSMMergePolicyFactory;
+import org.apache.hyracks.storage.am.rtree.frames.RTreePolicyType;
+
+public class LSMRTreeLocalResourceMetadataFactory extends ResourceFactory {
+
+    private static final long serialVersionUID = 1L;
+
+    protected final ITypeTraits[] typeTraits;
+    protected final IBinaryComparatorFactory[] rtreeCmpFactories;
+    protected final IBinaryComparatorFactory[] btreeCmpFactories;
+    protected final IPrimitiveValueProviderFactory[] valueProviderFactories;
+    protected final RTreePolicyType rtreePolicyType;
+    protected final ILinearizeComparatorFactory linearizeCmpFactory;
+    protected final ILSMMergePolicyFactory mergePolicyFactory;
+    protected final Map<String, String> mergePolicyProperties;
+    protected final int[] rtreeFields;
+    protected final int[] btreeFields;
+    protected final boolean isPointMBR;
+
+    public LSMRTreeLocalResourceMetadataFactory(ITypeTraits[] typeTraits, IBinaryComparatorFactory[] rtreeCmpFactories,
+            IBinaryComparatorFactory[] btreeCmpFactories, IPrimitiveValueProviderFactory[] valueProviderFactories,
+            RTreePolicyType rtreePolicyType, ILinearizeComparatorFactory linearizeCmpFactory, int datasetID,
+            ILSMMergePolicyFactory mergePolicyFactory, Map<String, String> mergePolicyProperties,
+            ITypeTraits[] filterTypeTraits, IBinaryComparatorFactory[] filterCmpFactories, int[] rtreeFields,
+            int[] btreeFields, int[] filterFields, boolean isPointMBR) {
+        super(datasetID, filterTypeTraits, filterCmpFactories, filterFields);
+        this.typeTraits = typeTraits;
+        this.rtreeCmpFactories = rtreeCmpFactories;
+        this.btreeCmpFactories = btreeCmpFactories;
+        this.valueProviderFactories = valueProviderFactories;
+        this.rtreePolicyType = rtreePolicyType;
+        this.linearizeCmpFactory = linearizeCmpFactory;
+        this.mergePolicyFactory = mergePolicyFactory;
+        this.mergePolicyProperties = mergePolicyProperties;
+        this.rtreeFields = rtreeFields;
+        this.btreeFields = btreeFields;
+        this.isPointMBR = isPointMBR;
+    }
+
+    @Override
+    public Resource resource(int partition) {
+        return new LSMRTreeLocalResourceMetadata(typeTraits, rtreeCmpFactories, btreeCmpFactories,
+                valueProviderFactories, rtreePolicyType, linearizeCmpFactory, datasetId, partition, mergePolicyFactory,
+                mergePolicyProperties, filterTypeTraits, filterCmpFactories, rtreeFields, btreeFields, filterFields,
+                isPointMBR);
+    }
+}

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactory.java
index 33e5bf2..8ab99fb 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactory.java
@@ -18,24 +18,24 @@
  */
 package org.apache.asterix.transaction.management.resource;
 
-import org.apache.asterix.common.api.ILocalResourceMetadata;
+import org.apache.asterix.common.transactions.IResourceFactory;
+import org.apache.asterix.common.transactions.Resource;
 import org.apache.hyracks.storage.common.file.ILocalResourceFactory;
 import org.apache.hyracks.storage.common.file.LocalResource;
 
 public class PersistentLocalResourceFactory implements ILocalResourceFactory {
 
-    private final ILocalResourceMetadata localResourceMetadata;
+    private final IResourceFactory applicationResourceFactory;
     private final int resourceType;
 
-    public PersistentLocalResourceFactory(ILocalResourceMetadata localResourceMetadata, int resourceType) {
-        this.localResourceMetadata = localResourceMetadata;
+    public PersistentLocalResourceFactory(IResourceFactory applicationResourceFactory, int resourceType) {
+        this.applicationResourceFactory = applicationResourceFactory;
         this.resourceType = resourceType;
     }
 
     @Override
-    public LocalResource createLocalResource(long resourceId, String resourceName, int partition, int storageVersion,
-            String resourcePath) {
-        return new LocalResource(resourceId, resourceName, partition, resourcePath, resourceType, storageVersion,
-                localResourceMetadata);
+    public LocalResource createLocalResource(long id, String name, int version, int partition) {
+        Resource resource = applicationResourceFactory.resource(partition);
+        return new LocalResource(id, name, resourceType, version, resource);
     }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactoryProvider.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactoryProvider.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactoryProvider.java
index 75d0769..c306699 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactoryProvider.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceFactoryProvider.java
@@ -18,23 +18,23 @@
  */
 package org.apache.asterix.transaction.management.resource;
 
-import org.apache.asterix.common.api.ILocalResourceMetadata;
+import org.apache.asterix.common.transactions.IResourceFactory;
 import org.apache.hyracks.storage.common.file.ILocalResourceFactory;
 import org.apache.hyracks.storage.common.file.ILocalResourceFactoryProvider;
 
 public class PersistentLocalResourceFactoryProvider implements ILocalResourceFactoryProvider {
 
     private static final long serialVersionUID = 1L;
-    private final ILocalResourceMetadata localResourceMetadata;
-    private final int resourceType;
+    private final IResourceFactory factory;
+    private final int type;
 
-    public PersistentLocalResourceFactoryProvider(ILocalResourceMetadata localResourceMetadata, int resourceType) {
-        this.localResourceMetadata = localResourceMetadata;
-        this.resourceType = resourceType;
+    public PersistentLocalResourceFactoryProvider(IResourceFactory factory, int type) {
+        this.factory = factory;
+        this.type = type;
     }
 
     @Override
     public ILocalResourceFactory getLocalResourceFactory() {
-        return new PersistentLocalResourceFactory(localResourceMetadata, resourceType);
+        return new PersistentLocalResourceFactory(factory, type);
     }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
index 006eac7..9ebb8e7 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepository.java
@@ -33,6 +33,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Set;
 import java.util.SortedMap;
 import java.util.concurrent.ConcurrentHashMap;
@@ -47,10 +48,12 @@ import org.apache.asterix.common.replication.IReplicationManager;
 import org.apache.asterix.common.utils.StoragePathUtil;
 import org.apache.commons.io.FileUtils;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.io.FileReference;
 import org.apache.hyracks.api.io.IODeviceHandle;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationExecutionType;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationJobType;
 import org.apache.hyracks.api.replication.IReplicationJob.ReplicationOperation;
+import org.apache.hyracks.control.nc.io.IOManager;
 import org.apache.hyracks.storage.am.common.frames.LIFOMetaDataFrame;
 import org.apache.hyracks.storage.common.file.ILocalResourceRepository;
 import org.apache.hyracks.storage.common.file.LocalResource;
@@ -64,13 +67,14 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
     public static final String METADATA_FILE_NAME = ".metadata";
     // Private constants
     private static final Logger LOGGER = Logger.getLogger(PersistentLocalResourceRepository.class.getName());
-    private static final String STORAGE_METADATA_DIRECTORY = "asterix_root_metadata";
-    private static final String STORAGE_METADATA_FILE_NAME_PREFIX = ".asterix_root_metadata";
+    private static final String STORAGE_METADATA_DIRECTORY = "root_metadata";
+    private static final String STORAGE_METADATA_FILE_NAME_PREFIX = ".root_metadata";
     private static final long STORAGE_LOCAL_RESOURCE_ID = -4321;
     private static final int MAX_CACHED_RESOURCES = 1000;
-    private static final FilenameFilter METADATA_FILES_FILTER =
-            (File dir, String name) -> name.equalsIgnoreCase(METADATA_FILE_NAME);
+    private static final FilenameFilter METADATA_FILES_FILTER = (File dir, String name) -> name.equalsIgnoreCase(
+            METADATA_FILE_NAME);
     // Finals
+    private final IOManager ioManager;
     private final String[] mountPoints;
     private final String nodeId;
     private final Cache<String, LocalResource> resourceCache;
@@ -83,13 +87,14 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
     private IReplicationManager replicationManager;
     private Set<Integer> nodeInactivePartitions;
 
-    public PersistentLocalResourceRepository(List<IODeviceHandle> devices, String nodeId,
+    public PersistentLocalResourceRepository(IOManager ioManager, List<IODeviceHandle> devices, String nodeId,
             AsterixMetadataProperties metadataProperties) throws HyracksDataException {
+        this.ioManager = ioManager;
         mountPoints = new String[devices.size()];
         this.nodeId = nodeId;
         this.clusterPartitions = metadataProperties.getClusterPartitions();
         for (int i = 0; i < mountPoints.length; i++) {
-            String mountPoint = devices.get(i).getPath().getPath();
+            String mountPoint = devices.get(i).getMount().getPath();
             File mountPointDir = new File(mountPoint);
             if (!mountPointDir.exists()) {
                 throw new HyracksDataException(mountPointDir.getAbsolutePath() + " doesn't exist.");
@@ -112,13 +117,16 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
         }
     }
 
-    private static String getStorageMetadataDirPath(String mountPoint, String nodeId, int ioDeviceId) {
-        return mountPoint + STORAGE_METADATA_DIRECTORY + File.separator + nodeId + "_" + "iodevice" + ioDeviceId;
-    }
-
-    private static File getStorageMetadataBaseDir(File storageMetadataFile) {
-        //STORAGE_METADATA_DIRECTORY / Node Id / STORAGE_METADATA_FILE_NAME_PREFIX
-        return storageMetadataFile.getParentFile().getParentFile();
+    @Override
+    public String toString() {
+        StringBuilder aString = new StringBuilder().append(PersistentLocalResourceRepository.class.getSimpleName())
+                .append(Character.LINE_SEPARATOR).append(ioManager.getClass().getSimpleName()).append(':').append(
+                        Character.LINE_SEPARATOR).append(ioManager.toString()).append(Character.LINE_SEPARATOR).append(
+                                "Cached Resources:").append(Character.LINE_SEPARATOR);
+        for (Entry<String, LocalResource> pair : resourceCache.asMap().entrySet()) {
+            aString.append(pair.getKey()).append("->").append(pair.getValue()).append(Character.LINE_SEPARATOR);
+        }
+        return aString.toString();
     }
 
     public void initializeNewUniverse(String storageRootDirName) throws HyracksDataException {
@@ -134,8 +142,8 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
          * This file could potentially store more information, otherwise no need to keep it.
          */
         for (int i = 0; i < mountPoints.length; i++) {
-            File storageMetadataFile = getStorageMetadataFile(mountPoints[i], nodeId, i);
-            File storageMetadataDir = storageMetadataFile.getParentFile();
+            FileReference storageMetadataFile = getStorageMetadataFile(ioManager, nodeId, i);
+            File storageMetadataDir = storageMetadataFile.getFile().getParentFile();
             //make dirs for the storage metadata file
             boolean success = storageMetadataDir.mkdirs();
             if (!success) {
@@ -156,8 +164,7 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
             }
 
             LocalResource rootLocalResource = new LocalResource(STORAGE_LOCAL_RESOURCE_ID,
-                    storageMetadataFile.getAbsolutePath(), 0, storageMetadataFile.getAbsolutePath(), 0,
-                    LIFOMetaDataFrame.VERSION, storageRootDirPath);
+                    storageMetadataFile.getRelativePath(), 0, LIFOMetaDataFrame.VERSION, storageRootDirPath);
             insert(rootLocalResource);
             LOGGER.log(Level.INFO, "created the root-metadata-file: " + storageMetadataFile.getAbsolutePath());
         }
@@ -165,13 +172,13 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
     }
 
     @Override
-    public LocalResource getResourceByPath(String path) throws HyracksDataException {
-        LocalResource resource = resourceCache.getIfPresent(path);
+    public LocalResource get(String relativePath) throws HyracksDataException {
+        LocalResource resource = resourceCache.getIfPresent(relativePath);
         if (resource == null) {
-            File resourceFile = getLocalResourceFileByName(path);
-            if (resourceFile.exists()) {
-                resource = readLocalResource(resourceFile);
-                resourceCache.put(path, resource);
+            FileReference resourceFile = getLocalResourceFileByName(ioManager, relativePath);
+            if (resourceFile.getFile().exists()) {
+                resource = readLocalResource(resourceFile.getFile());
+                resourceCache.put(relativePath, resource);
             }
         }
         return resource;
@@ -179,18 +186,19 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
 
     @Override
     public synchronized void insert(LocalResource resource) throws HyracksDataException {
-        File resourceFile = new File(getFileName(resource.getResourcePath(), resource.getResourceId()));
-        if (resourceFile.exists()) {
+        String relativePath = getFileName(resource.getPath(), resource.getId());
+        FileReference resourceFile = ioManager.getFileRef(relativePath, true);
+        if (resourceFile.getFile().exists()) {
             throw new HyracksDataException("Duplicate resource: " + resourceFile.getAbsolutePath());
         } else {
-            resourceFile.getParentFile().mkdirs();
+            resourceFile.getFile().getParentFile().mkdirs();
         }
 
-        if (resource.getResourceId() != STORAGE_LOCAL_RESOURCE_ID) {
-            resourceCache.put(resource.getResourcePath(), resource);
+        if (resource.getId() != STORAGE_LOCAL_RESOURCE_ID) {
+            resourceCache.put(resource.getPath(), resource);
         }
 
-        try (FileOutputStream fos = new FileOutputStream(resourceFile);
+        try (FileOutputStream fos = new FileOutputStream(resourceFile.getFile());
                 ObjectOutputStream oosToFos = new ObjectOutputStream(fos)) {
             oosToFos.writeObject(resource);
             oosToFos.flush();
@@ -199,21 +207,22 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
         }
 
         //if replication enabled, send resource metadata info to remote nodes
-        if (isReplicationEnabled && resource.getResourceId() != STORAGE_LOCAL_RESOURCE_ID) {
-            String filePath = getFileName(resource.getResourcePath(), resource.getResourceId());
+        if (isReplicationEnabled && resource.getId() != STORAGE_LOCAL_RESOURCE_ID) {
+            String filePath = getFileName(resource.getPath(), resource.getId());
             createReplicationJob(ReplicationOperation.REPLICATE, filePath);
         }
     }
 
     @Override
-    public synchronized void deleteResourceByPath(String resourcePath) throws HyracksDataException {
-        File resourceFile = getLocalResourceFileByName(resourcePath);
-        if (resourceFile.exists()) {
+    public synchronized void delete(String relativePath) throws HyracksDataException {
+        FileReference resourceFile = getLocalResourceFileByName(ioManager, relativePath);
+        if (resourceFile.getFile().exists()) {
             resourceFile.delete();
-            resourceCache.invalidate(resourcePath);
+            resourceCache.invalidate(relativePath);
 
             //if replication enabled, delete resource from remote replicas
-            if (isReplicationEnabled && !resourceFile.getName().startsWith(STORAGE_METADATA_FILE_NAME_PREFIX)) {
+            if (isReplicationEnabled && !resourceFile.getFile().getName().startsWith(
+                    STORAGE_METADATA_FILE_NAME_PREFIX)) {
                 createReplicationJob(ReplicationOperation.DELETE, resourceFile.getAbsolutePath());
             }
         } else {
@@ -221,8 +230,10 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
         }
     }
 
-    private static File getLocalResourceFileByName(String resourcePath) {
-        return new File(resourcePath + File.separator + METADATA_FILE_NAME);
+    private static FileReference getLocalResourceFileByName(IOManager ioManager, String resourcePath)
+            throws HyracksDataException {
+        String fileName = resourcePath + File.separator + METADATA_FILE_NAME;
+        return ioManager.getFileRef(fileName, true);
     }
 
     public Map<Long, LocalResource> loadAndGetAllResources() throws HyracksDataException {
@@ -230,7 +241,7 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
         //This could be fixed by traversing all resources on disk until the required resource is found.
         Map<Long, LocalResource> resourcesMap = new HashMap<>();
         for (int i = 0; i < mountPoints.length; i++) {
-            File storageRootDir = getStorageRootDirectoryIfExists(mountPoints[i], nodeId, i);
+            File storageRootDir = getStorageRootDirectoryIfExists(ioManager, nodeId, i);
             if (storageRootDir == null) {
                 continue;
             }
@@ -265,18 +276,18 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
             if (metadataFiles != null) {
                 for (File metadataFile : metadataFiles) {
                     LocalResource localResource = readLocalResource(metadataFile);
-                    resourcesMap.put(localResource.getResourceId(), localResource);
+                    resourcesMap.put(localResource.getId(), localResource);
                 }
             }
         }
     }
 
     @Override
-    public long getMaxResourceID() throws HyracksDataException {
+    public long maxId() throws HyracksDataException {
         long maxResourceId = 0;
 
         for (int i = 0; i < mountPoints.length; i++) {
-            File storageRootDir = getStorageRootDirectoryIfExists(mountPoints[i], nodeId, i);
+            File storageRootDir = getStorageRootDirectoryIfExists(ioManager, nodeId, i);
             if (storageRootDir == null) {
                 continue;
             }
@@ -316,7 +327,7 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
             if (metadataFiles != null) {
                 for (File metadataFile : metadataFiles) {
                     LocalResource localResource = readLocalResource(metadataFile);
-                    maxResourceId = Math.max(maxResourceId, localResource.getResourceId());
+                    maxResourceId = Math.max(maxResourceId, localResource.getId());
                 }
             }
         }
@@ -388,14 +399,14 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
      */
     public void deleteStorageData(boolean deleteStorageMetadata) throws IOException {
         for (int i = 0; i < mountPoints.length; i++) {
-            File storageDir = getStorageRootDirectoryIfExists(mountPoints[i], nodeId, i);
+            File storageDir = getStorageRootDirectoryIfExists(ioManager, nodeId, i);
             if (storageDir != null && storageDir.isDirectory()) {
                 FileUtils.deleteDirectory(storageDir);
             }
             if (deleteStorageMetadata) {
                 //delete the metadata root directory
-                File storageMetadataFile = getStorageMetadataFile(mountPoints[i], nodeId, i);
-                File storageMetadataDir = getStorageMetadataBaseDir(storageMetadataFile);
+                FileReference storageMetadataFile = getStorageMetadataFile(ioManager, nodeId, i);
+                File storageMetadataDir = storageMetadataFile.getFile().getParentFile().getParentFile();
                 if (storageMetadataDir.exists() && storageMetadataDir.isDirectory()) {
                     FileUtils.deleteDirectory(storageMetadataDir);
                 }
@@ -409,10 +420,12 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
      * @param ioDeviceId
      * @return A file reference to the storage metadata file.
      */
-    private static File getStorageMetadataFile(String mountPoint, String nodeId, int ioDeviceId) {
-        String storageMetadataFileName = getStorageMetadataDirPath(mountPoint, nodeId, ioDeviceId) + File.separator
+    private static FileReference getStorageMetadataFile(IOManager ioManager, String nodeId,
+            int ioDeviceId) {
+        String storageMetadataFileName = STORAGE_METADATA_DIRECTORY + File.separator + nodeId + "_" + "iodevice"
+                + ioDeviceId + File.separator
                 + STORAGE_METADATA_FILE_NAME_PREFIX;
-        return new File(storageMetadataFileName);
+        return new FileReference(ioManager.getIODevices().get(ioDeviceId), storageMetadataFileName);
     }
 
     /**
@@ -422,13 +435,14 @@ public class PersistentLocalResourceRepository implements ILocalResourceReposito
      * @return A file reference to the storage root directory if exists, otherwise null.
      * @throws HyracksDataException
      */
-    public static File getStorageRootDirectoryIfExists(String mountPoint, String nodeId, int ioDeviceId)
+    public static File getStorageRootDirectoryIfExists(IOManager ioManager, String nodeId,
+            int ioDeviceId)
             throws HyracksDataException {
         File storageRootDir = null;
-        File storageMetadataFile = getStorageMetadataFile(mountPoint, nodeId, ioDeviceId);
-        if (storageMetadataFile.exists()) {
-            LocalResource rootLocalResource = readLocalResource(storageMetadataFile);
-            String storageRootDirPath = (String) rootLocalResource.getResourceObject();
+        FileReference storageMetadataFile = getStorageMetadataFile(ioManager, nodeId, ioDeviceId);
+        if (storageMetadataFile.getFile().exists()) {
+            LocalResource rootLocalResource = readLocalResource(storageMetadataFile.getFile());
+            String storageRootDirPath = (String) rootLocalResource.getResource();
             Path path = Paths.get(storageRootDirPath);
             if (Files.exists(path)) {
                 storageRootDir = new File(storageRootDirPath);

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
index d62abc5..1c28e17 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/resource/PersistentLocalResourceRepositoryFactory.java
@@ -20,16 +20,16 @@ package org.apache.asterix.transaction.management.resource;
 
 import org.apache.asterix.common.config.AsterixMetadataProperties;
 import org.apache.hyracks.api.exceptions.HyracksDataException;
-import org.apache.hyracks.api.io.IIOManager;
+import org.apache.hyracks.control.nc.io.IOManager;
 import org.apache.hyracks.storage.common.file.ILocalResourceRepository;
 import org.apache.hyracks.storage.common.file.ILocalResourceRepositoryFactory;
 
 public class PersistentLocalResourceRepositoryFactory implements ILocalResourceRepositoryFactory {
-    private final IIOManager ioManager;
+    private final IOManager ioManager;
     private final String nodeId;
     private final AsterixMetadataProperties metadataProperties;
 
-    public PersistentLocalResourceRepositoryFactory(IIOManager ioManager, String nodeId,
+    public PersistentLocalResourceRepositoryFactory(IOManager ioManager, String nodeId,
             AsterixMetadataProperties metadataProperties) {
         this.ioManager = ioManager;
         this.nodeId = nodeId;
@@ -38,6 +38,6 @@ public class PersistentLocalResourceRepositoryFactory implements ILocalResourceR
 
     @Override
     public ILocalResourceRepository createRepository() throws HyracksDataException {
-        return new PersistentLocalResourceRepository(ioManager.getIODevices(), nodeId, metadataProperties);
+        return new PersistentLocalResourceRepository(ioManager, ioManager.getIODevices(), nodeId, metadataProperties);
     }
 }

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
----------------------------------------------------------------------
diff --git a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
index 286fc16..1f8d698 100644
--- a/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
+++ b/asterixdb/asterix-transactions/src/main/java/org/apache/asterix/transaction/management/service/recovery/RecoveryManager.java
@@ -48,7 +48,6 @@ import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.apache.asterix.common.api.IDatasetLifecycleManager;
-import org.apache.asterix.common.api.ILocalResourceMetadata;
 import org.apache.asterix.common.cluster.ClusterPartition;
 import org.apache.asterix.common.config.AsterixMetadataProperties;
 import org.apache.asterix.common.config.ClusterProperties;
@@ -63,6 +62,7 @@ import org.apache.asterix.common.transactions.ILogRecord;
 import org.apache.asterix.common.transactions.IRecoveryManager;
 import org.apache.asterix.common.transactions.ITransactionContext;
 import org.apache.asterix.common.transactions.LogType;
+import org.apache.asterix.common.transactions.Resource;
 import org.apache.asterix.transaction.management.resource.PersistentLocalResourceRepository;
 import org.apache.asterix.transaction.management.service.logging.LogManager;
 import org.apache.asterix.transaction.management.service.transaction.TransactionManagementConstants;
@@ -214,7 +214,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
         int abortLogCount = 0;
         int jobId = -1;
 
-        Set<Integer> winnerJobSet = new HashSet<Integer>();
+        Set<Integer> winnerJobSet = new HashSet<>();
         jobId2WinnerEntitiesMap = new HashMap<>();
 
         //set log reader to the lowWaterMarkLsn
@@ -300,7 +300,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
         long LSN = -1;
         ILSMIndex index = null;
         LocalResource localResource = null;
-        ILocalResourceMetadata localResourceMetadata = null;
+        Resource localResourceMetadata = null;
         boolean foundWinner = false;
         JobEntityCommits jobEntityWinners = null;
 
@@ -308,7 +308,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
         IDatasetLifecycleManager datasetLifecycleManager = appRuntimeContext.getDatasetLifecycleManager();
 
         Map<Long, LocalResource> resourcesMap = localResourceRepository.loadAndGetAllResources();
-        Map<Long, Long> resourceId2MaxLSNMap = new HashMap<Long, Long>();
+        Map<Long, Long> resourceId2MaxLSNMap = new HashMap<>();
         TxnId tempKeyTxnId = new TxnId(-1, -1, -1, null, -1, false);
 
         ILogRecord logRecord = null;
@@ -361,20 +361,13 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
                                 //if index is not registered into IndexLifeCycleManager,
                                 //create the index using LocalMetadata stored in LocalResourceRepository
                                 //get partition path in this node
-                                String partitionIODevicePath =
-                                        localResourceRepository.getPartitionPath(localResource.getPartition());
-                                String resourceAbsolutePath =
-                                        partitionIODevicePath + File.separator + localResource.getResourceName();
-                                localResource.setResourcePath(resourceAbsolutePath);
-                                index = (ILSMIndex) datasetLifecycleManager.get(resourceAbsolutePath);
+                                localResourceMetadata = (Resource) localResource.getResource();
+                                index = (ILSMIndex) datasetLifecycleManager.get(localResource.getPath());
                                 if (index == null) {
                                     //#. create index instance and register to indexLifeCycleManager
-                                    localResourceMetadata = (ILocalResourceMetadata) localResource.getResourceObject();
-                                    index = localResourceMetadata.createIndexInstance(appRuntimeContext,
-                                            resourceAbsolutePath, localResource.getPartition(),
-                                            localResourceRepository.getIODeviceNum(localResource.getPartition()));
-                                    datasetLifecycleManager.register(resourceAbsolutePath, index);
-                                    datasetLifecycleManager.open(resourceAbsolutePath);
+                                    index = localResourceMetadata.createIndexInstance(appRuntimeContext, localResource);
+                                    datasetLifecycleManager.register(localResource.getPath(), index);
+                                    datasetLifecycleManager.open(localResource.getPath());
 
                                     //#. get maxDiskLastLSN
                                     ILSMIndex lsmIndex = index;
@@ -383,7 +376,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
                                                 ((AbstractLSMIOOperationCallback) lsmIndex.getIOOperationCallback())
                                                         .getComponentLSN(lsmIndex.getImmutableComponents());
                                     } catch (HyracksDataException e) {
-                                        datasetLifecycleManager.close(resourceAbsolutePath);
+                                        datasetLifecycleManager.close(localResource.getPath());
                                         throw e;
                                     }
 
@@ -419,7 +412,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
             //close all indexes
             Set<Long> resourceIdList = resourceId2MaxLSNMap.keySet();
             for (long r : resourceIdList) {
-                datasetLifecycleManager.close(resourcesMap.get(r).getResourcePath());
+                datasetLifecycleManager.close(resourcesMap.get(r).getPath());
             }
         }
     }
@@ -602,7 +595,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
             throw new FileNotFoundException("Checkpoint file is not found");
         }
 
-        List<CheckpointObject> checkpointObjectList = new ArrayList<CheckpointObject>();
+        List<CheckpointObject> checkpointObjectList = new ArrayList<>();
         for (File file : prevCheckpointFiles) {
             try (FileInputStream fis = new FileInputStream(file);
                     ObjectInputStream oisFromFis = new ObjectInputStream(fis)) {
@@ -720,7 +713,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
             LOGGER.info("collecting loser transaction's LSNs from " + firstLSN + " to " + lastLSN);
         }
 
-        Map<TxnId, List<Long>> jobLoserEntity2LSNsMap = new HashMap<TxnId, List<Long>>();
+        Map<TxnId, List<Long>> jobLoserEntity2LSNsMap = new HashMap<>();
         TxnId tempKeyTxnId = new TxnId(-1, -1, -1, null, -1, false);
         int updateLogCount = 0;
         int entityCommitLogCount = 0;
@@ -758,7 +751,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
                             if (undoLSNSet == null) {
                                 loserEntity = new TxnId(logJobId, logRecord.getDatasetId(), logRecord.getPKHashValue(),
                                         logRecord.getPKValue(), logRecord.getPKValueSize(), true);
-                                undoLSNSet = new LinkedList<Long>();
+                                undoLSNSet = new LinkedList<>();
                                 jobLoserEntity2LSNsMap.put(loserEntity, undoLSNSet);
                             }
                             undoLSNSet.add(currentLSN);
@@ -876,8 +869,10 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
 
     private static void redo(ILogRecord logRecord, IDatasetLifecycleManager datasetLifecycleManager) {
         try {
+            int datasetId = logRecord.getDatasetId();
+            long resourceId = logRecord.getResourceId();
             ILSMIndex index =
-                    (ILSMIndex) datasetLifecycleManager.getIndex(logRecord.getDatasetId(), logRecord.getResourceId());
+                    (ILSMIndex) datasetLifecycleManager.getIndex(datasetId, resourceId);
             ILSMIndexAccessor indexAccessor =
                     index.createAccessor(NoOpOperationCallback.INSTANCE, NoOpOperationCallback.INSTANCE);
             if (logRecord.getNewOp() == IndexOperation.INSERT.ordinal()) {
@@ -888,7 +883,6 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
                 throw new IllegalStateException("Unsupported OperationType: " + logRecord.getNewOp());
             }
         } catch (Exception e) {
-            e.printStackTrace();
             throw new IllegalStateException("Failed to redo", e);
         }
     }
@@ -896,8 +890,8 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
     private class JobEntityCommits {
         private static final String PARTITION_FILE_NAME_SEPARATOR = "_";
         private final int jobId;
-        private final Set<TxnId> cachedEntityCommitTxns = new HashSet<TxnId>();
-        private final List<File> jobEntitCommitOnDiskPartitionsFiles = new ArrayList<File>();
+        private final Set<TxnId> cachedEntityCommitTxns = new HashSet<>();
+        private final List<File> jobEntitCommitOnDiskPartitionsFiles = new ArrayList<>();
         //a flag indicating whether all the the commits for this jobs have been added.
         private boolean preparedForSearch = false;
         private TxnId winnerEntity = null;
@@ -975,7 +969,7 @@ public class RecoveryManager implements IRecoveryManager, ILifeCycleComponent {
          * @return partitions that have a max LSN > logLSN
          */
         public ArrayList<File> getCandidiatePartitions(long logLSN) {
-            ArrayList<File> candidiatePartitions = new ArrayList<File>();
+            ArrayList<File> candidiatePartitions = new ArrayList<>();
             for (File partition : jobEntitCommitOnDiskPartitionsFiles) {
                 String partitionName = partition.getName();
                 //entity commit log must come after the update log, therefore, consider only partitions with max LSN > logLSN

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/properties/FileSplitDomain.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/properties/FileSplitDomain.java b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/properties/FileSplitDomain.java
index 4afcda8..9347947 100644
--- a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/properties/FileSplitDomain.java
+++ b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/properties/FileSplitDomain.java
@@ -18,7 +18,7 @@
  */
 package org.apache.hyracks.algebricks.core.algebra.properties;
 
-import org.apache.hyracks.dataflow.std.file.FileSplit;
+import org.apache.hyracks.api.io.FileSplit;
 
 public class FileSplitDomain implements INodeDomain {
 
@@ -44,7 +44,7 @@ public class FileSplitDomain implements INodeDomain {
             } else {
                 sb.append(", ");
             }
-            sb.append(fs.getNodeName() + ":" + fs.getLocalFile());
+            sb.append(fs.getNodeName() + ":" + fs.getPath());
         }
         sb.append(']');
         return sb.toString();

http://git-wip-us.apache.org/repos/asf/asterixdb/blob/163c3be3/hyracks-fullstack/algebricks/algebricks-examples/piglet-example/src/main/java/org/apache/hyracks/algebricks/examples/piglet/metadata/FileSplitUtils.java
----------------------------------------------------------------------
diff --git a/hyracks-fullstack/algebricks/algebricks-examples/piglet-example/src/main/java/org/apache/hyracks/algebricks/examples/piglet/metadata/FileSplitUtils.java b/hyracks-fullstack/algebricks/algebricks-examples/piglet-example/src/main/java/org/apache/hyracks/algebricks/examples/piglet/metadata/FileSplitUtils.java
index fe3e1d5..d01bae5 100644
--- a/hyracks-fullstack/algebricks/algebricks-examples/piglet-example/src/main/java/org/apache/hyracks/algebricks/examples/piglet/metadata/FileSplitUtils.java
+++ b/hyracks-fullstack/algebricks/algebricks-examples/piglet-example/src/main/java/org/apache/hyracks/algebricks/examples/piglet/metadata/FileSplitUtils.java
@@ -20,8 +20,7 @@ package org.apache.hyracks.algebricks.examples.piglet.metadata;
 
 import java.io.File;
 
-import org.apache.hyracks.api.io.FileReference;
-import org.apache.hyracks.dataflow.std.file.FileSplit;
+import org.apache.hyracks.api.io.FileSplit;
 
 public class FileSplitUtils {
     public static FileSplit[] parseFileSplits(String fileSplits) {
@@ -33,7 +32,7 @@ public class FileSplitUtils {
             if (idx < 0) {
                 throw new IllegalArgumentException("File split " + s + " not well formed");
             }
-            fSplits[i] = new FileSplit(s.substring(0, idx), new FileReference(new File(s.substring(idx + 1))));
+            fSplits[i] = new FileSplit(s.substring(0, idx), new File(s.substring(idx + 1)).getAbsolutePath(), false);
         }
         return fSplits;
     }


Mime
View raw message