--- a/net/minecraft/world/level/material/FlowingFluid.java
+++ b/net/minecraft/world/level/material/FlowingFluid.java
@@ -45,6 +_,48 @@
     });
     private final Map<FluidState, VoxelShape> shapes = Maps.newIdentityHashMap();
 
+    // Paper start - fluid method optimisations
+    private FluidState sourceFalling;
+    private FluidState sourceNotFalling;
+
+    private static final int TOTAL_FLOWING_STATES = FALLING.getPossibleValues().size() * LEVEL.getPossibleValues().size();
+    private static final int MIN_LEVEL = LEVEL.getPossibleValues().stream().sorted().findFirst().get().intValue();
+
+    // index = (falling ? 1 : 0) + level*2
+    private FluidState[] flowingLookUp;
+    private volatile boolean init;
+
+    private static final int COLLISION_OCCLUSION_CACHE_SIZE = 2048;
+    private static final ThreadLocal<ca.spottedleaf.moonrise.patches.collisions.util.FluidOcclusionCacheKey[]> COLLISION_OCCLUSION_CACHE = ThreadLocal.withInitial(() -> new ca.spottedleaf.moonrise.patches.collisions.util.FluidOcclusionCacheKey[COLLISION_OCCLUSION_CACHE_SIZE]);
+
+
+    /**
+     * Due to init order, we need to use callbacks to initialise our state
+     */
+    private void init() {
+        synchronized (this) {
+            if (this.init) {
+                return;
+            }
+            this.flowingLookUp = new FluidState[TOTAL_FLOWING_STATES];
+            final FluidState defaultFlowState = this.getFlowing().defaultFluidState();
+            for (int i = 0; i < TOTAL_FLOWING_STATES; ++i) {
+                final int falling = i & 1;
+                final int level = (i >>> 1) + MIN_LEVEL;
+
+                this.flowingLookUp[i] = defaultFlowState.setValue(FALLING, falling == 1 ? Boolean.TRUE : Boolean.FALSE)
+                    .setValue(LEVEL, Integer.valueOf(level));
+            }
+
+            final FluidState defaultFallState = this.getSource().defaultFluidState();
+            this.sourceFalling = defaultFallState.setValue(FALLING, Boolean.TRUE);
+            this.sourceNotFalling = defaultFallState.setValue(FALLING, Boolean.FALSE);
+
+            this.init = true;
+        }
+    }
+    // Paper end - fluid method optimisations
+
     @Override
     protected void createFluidStateDefinition(StateDefinition.Builder<Fluid, FluidState> builder) {
         builder.add(FALLING);
@@ -118,6 +_,15 @@
                 FluidState newLiquid = this.getNewLiquid(level, blockPos, blockState1);
                 Fluid type = newLiquid.getType();
                 if (fluidState1.canBeReplacedWith(level, blockPos, type, Direction.DOWN) && canHoldSpecificFluid(level, blockPos, blockState1, type)) {
+                    // CraftBukkit start
+                    org.bukkit.block.Block source = org.bukkit.craftbukkit.block.CraftBlock.at(level, pos);
+                    org.bukkit.event.block.BlockFromToEvent event = new org.bukkit.event.block.BlockFromToEvent(source, org.bukkit.block.BlockFace.DOWN);
+                    level.getCraftServer().getPluginManager().callEvent(event);
+
+                    if (event.isCancelled()) {
+                        return;
+                    }
+                    // CraftBukkit end
                     this.spreadTo(level, blockPos, blockState1, Direction.DOWN, newLiquid);
                     if (this.sourceNeighborCount(level, pos) >= 3) {
                         this.spreadToSides(level, pos, fluidState, blockState);
@@ -146,7 +_,18 @@
                 Direction direction = entry.getKey();
                 FluidState fluidState1 = entry.getValue();
                 BlockPos blockPos = pos.relative(direction);
-                this.spreadTo(level, blockPos, level.getBlockState(blockPos), direction, fluidState1);
+                final BlockState blockStateIfLoaded = level.getBlockStateIfLoaded(blockPos); // Paper - Prevent chunk loading from fluid flowing
+                if (blockStateIfLoaded == null) continue; // Paper - Prevent chunk loading from fluid flowing
+                // CraftBukkit start
+                org.bukkit.block.Block source = org.bukkit.craftbukkit.block.CraftBlock.at(level, pos);
+                org.bukkit.event.block.BlockFromToEvent event = new org.bukkit.event.block.BlockFromToEvent(source, org.bukkit.craftbukkit.block.CraftBlock.notchToBlockFace(direction));
+                level.getCraftServer().getPluginManager().callEvent(event);
+
+                if (event.isCancelled()) {
+                    continue;
+                }
+                // CraftBukkit end
+                this.spreadTo(level, blockPos, blockStateIfLoaded, direction, fluidState1); // Paper - Prevent chunk loading from fluid flowing
             }
         }
     }
@@ -158,7 +_,8 @@
 
         for (Direction direction : Direction.Plane.HORIZONTAL) {
             BlockPos blockPos = mutableBlockPos.setWithOffset(pos, direction);
-            BlockState blockState = level.getBlockState(blockPos);
+            BlockState blockState = level.getBlockStateIfLoaded(blockPos); // Paper - Prevent chunk loading from fluid flowing
+            if (blockState == null) continue; // Paper - Prevent chunk loading from fluid flowing
             FluidState fluidState = blockState.getFluidState();
             if (fluidState.getType().isSame(this) && canPassThroughWall(direction, level, pos, state, blockPos, blockState)) {
                 if (fluidState.isSource()) {
@@ -188,61 +_,71 @@
         }
     }
 
-    private static boolean canPassThroughWall(
-        Direction direction, BlockGetter level, BlockPos pos, BlockState state, BlockPos spreadPos, BlockState spreadState
-    ) {
-        VoxelShape collisionShape = spreadState.getCollisionShape(level, spreadPos);
-        if (collisionShape == Shapes.block()) {
+    // Paper start - fluid method optimisations
+    private static boolean canPassThroughWall(final Direction direction, final BlockGetter level,
+                                              final BlockPos fromPos, final BlockState fromState,
+                                              final BlockPos toPos, final BlockState toState) {
+        if (((ca.spottedleaf.moonrise.patches.collisions.block.CollisionBlockState)fromState).moonrise$emptyCollisionShape() & ((ca.spottedleaf.moonrise.patches.collisions.block.CollisionBlockState)toState).moonrise$emptyCollisionShape()) {
+            // don't even try to cache simple cases
+            return true;
+        }
+
+        if (((ca.spottedleaf.moonrise.patches.collisions.block.CollisionBlockState)fromState).moonrise$occludesFullBlock() | ((ca.spottedleaf.moonrise.patches.collisions.block.CollisionBlockState)toState).moonrise$occludesFullBlock()) {
+            // don't even try to cache simple cases
             return false;
-        } else {
-            VoxelShape collisionShape1 = state.getCollisionShape(level, pos);
-            if (collisionShape1 == Shapes.block()) {
-                return false;
-            } else if (collisionShape1 == Shapes.empty() && collisionShape == Shapes.empty()) {
-                return true;
-            } else {
-                Object2ByteLinkedOpenHashMap<FlowingFluid.BlockStatePairKey> map;
-                if (!state.getBlock().hasDynamicShape() && !spreadState.getBlock().hasDynamicShape()) {
-                    map = OCCLUSION_CACHE.get();
-                } else {
-                    map = null;
-                }
-
-                FlowingFluid.BlockStatePairKey blockStatePairKey;
-                if (map != null) {
-                    blockStatePairKey = new FlowingFluid.BlockStatePairKey(state, spreadState, direction);
-                    byte andMoveToFirst = map.getAndMoveToFirst(blockStatePairKey);
-                    if (andMoveToFirst != 127) {
-                        return andMoveToFirst != 0;
-                    }
-                } else {
-                    blockStatePairKey = null;
-                }
-
-                boolean flag = !Shapes.mergedFaceOccludes(collisionShape1, collisionShape, direction);
-                if (map != null) {
-                    if (map.size() == 200) {
-                        map.removeLastByte();
-                    }
-
-                    map.putAndMoveToFirst(blockStatePairKey, (byte)(flag ? 1 : 0));
-                }
-
-                return flag;
+        }
+
+        final ca.spottedleaf.moonrise.patches.collisions.util.FluidOcclusionCacheKey[] cache = ((ca.spottedleaf.moonrise.patches.collisions.block.CollisionBlockState)fromState).moonrise$hasCache() & ((ca.spottedleaf.moonrise.patches.collisions.block.CollisionBlockState)toState).moonrise$hasCache() ?
+            COLLISION_OCCLUSION_CACHE.get() : null;
+
+        final int keyIndex
+            = (((ca.spottedleaf.moonrise.patches.collisions.block.CollisionBlockState)fromState).moonrise$uniqueId1() ^ ((ca.spottedleaf.moonrise.patches.collisions.block.CollisionBlockState)toState).moonrise$uniqueId2() ^ ((ca.spottedleaf.moonrise.patches.collisions.util.CollisionDirection)(Object)direction).moonrise$uniqueId())
+            & (COLLISION_OCCLUSION_CACHE_SIZE - 1);
+
+        if (cache != null) {
+            final ca.spottedleaf.moonrise.patches.collisions.util.FluidOcclusionCacheKey cached = cache[keyIndex];
+            if (cached != null && cached.first() == fromState && cached.second() == toState && cached.direction() == direction) {
+                return cached.result();
             }
         }
+
+        final VoxelShape shape1 = fromState.getCollisionShape(level, fromPos);
+        final VoxelShape shape2 = toState.getCollisionShape(level, toPos);
+
+        final boolean result = !Shapes.mergedFaceOccludes(shape1, shape2, direction);
+
+        if (cache != null) {
+            // we can afford to replace in-use keys more often due to the excessive caching the collision patch does in mergedFaceOccludes
+            cache[keyIndex] = new ca.spottedleaf.moonrise.patches.collisions.util.FluidOcclusionCacheKey(fromState, toState, direction, result);
+        }
+
+        return result;
     }
+    // Paper end - fluid method optimisations
+
 
     public abstract Fluid getFlowing();
 
     public FluidState getFlowing(int level, boolean falling) {
-        return this.getFlowing().defaultFluidState().setValue(LEVEL, Integer.valueOf(level)).setValue(FALLING, Boolean.valueOf(falling));
+        // Paper start - fluid method optimisations
+        final int amount = level;
+        if (!this.init) {
+            this.init();
+        }
+        final int index = (falling ? 1 : 0) | ((amount - MIN_LEVEL) << 1);
+        return this.flowingLookUp[index];
+        // Paper end - fluid method optimisations
     }
 
     public abstract Fluid getSource();
 
     public FluidState getSource(boolean falling) {
-        return this.getSource().defaultFluidState().setValue(FALLING, Boolean.valueOf(falling));
+        // Paper start - fluid method optimisations
+        if (!this.init) {
+            this.init();
+        }
+        return falling ? this.sourceFalling : this.sourceNotFalling;
+        // Paper end - fluid method optimisations
     }
 
     protected abstract boolean canConvertToSource(ServerLevel level);
@@ -252,13 +_,14 @@
             liquidBlockContainer.placeLiquid(level, pos, blockState, fluidState);
         } else {
             if (!blockState.isAir()) {
-                this.beforeDestroyingBlock(level, pos, blockState);
+                this.beforeDestroyingBlock(level, pos, blockState, pos.relative(direction.getOpposite())); // Paper - Add BlockBreakBlockEvent
             }
 
             level.setBlock(pos, fluidState.createLegacyBlock(), 3);
         }
     }
 
+    protected void beforeDestroyingBlock(LevelAccessor level, BlockPos pos, BlockState state, BlockPos source) { beforeDestroyingBlock(level, pos, state); } // Paper - Add BlockBreakBlockEvent
     protected abstract void beforeDestroyingBlock(LevelAccessor level, BlockPos pos, BlockState state);
 
     protected int getSlopeDistance(LevelReader level, BlockPos pos, int depth, Direction direction, BlockState state, FlowingFluid.SpreadContext spreadContext) {
@@ -267,7 +_,8 @@
         for (Direction direction1 : Direction.Plane.HORIZONTAL) {
             if (direction1 != direction) {
                 BlockPos blockPos = pos.relative(direction1);
-                BlockState blockState = spreadContext.getBlockState(blockPos);
+                BlockState blockState = spreadContext.getBlockStateIfLoaded(blockPos); // Paper - Prevent chunk loading from fluid flowing
+                if (blockState == null) continue; // Paper - Prevent chunk loading from fluid flowing
                 FluidState fluidState = blockState.getFluidState();
                 if (this.canPassThrough(level, this.getFlowing(), pos, state, direction1, blockPos, blockState, fluidState)) {
                     if (spreadContext.isHole(blockPos)) {
@@ -334,7 +_,8 @@
 
         for (Direction direction : Direction.Plane.HORIZONTAL) {
             BlockPos blockPos = pos.relative(direction);
-            BlockState blockState = level.getBlockState(blockPos);
+            BlockState blockState = level.getBlockStateIfLoaded(blockPos); // Paper - Prevent chunk loading from fluid flowing
+            if (blockState == null) continue; // Paper - Prevent chunk loading from fluid flowing
             FluidState fluidState = blockState.getFluidState();
             if (this.canMaybePassThrough(level, pos, state, direction, blockPos, blockState, fluidState)) {
                 FluidState newLiquid = this.getNewLiquid(level, blockPos, blockState);
@@ -405,10 +_,24 @@
             if (newLiquid.isEmpty()) {
                 fluidState = newLiquid;
                 blockState = Blocks.AIR.defaultBlockState();
+                // CraftBukkit start
+                org.bukkit.event.block.FluidLevelChangeEvent event = org.bukkit.craftbukkit.event.CraftEventFactory.callFluidLevelChangeEvent(level, pos, blockState);
+                if (event.isCancelled()) {
+                    return;
+                }
+                blockState = ((org.bukkit.craftbukkit.block.data.CraftBlockData) event.getNewData()).getState();
+                // CraftBukkit end
                 level.setBlock(pos, blockState, 3);
             } else if (!newLiquid.equals(fluidState)) {
                 fluidState = newLiquid;
                 blockState = newLiquid.createLegacyBlock();
+                // CraftBukkit start
+                org.bukkit.event.block.FluidLevelChangeEvent event = org.bukkit.craftbukkit.event.CraftEventFactory.callFluidLevelChangeEvent(level, pos, blockState);
+                if (event.isCancelled()) {
+                    return;
+                }
+                blockState = ((org.bukkit.craftbukkit.block.data.CraftBlockData) event.getNewData()).getState();
+                // CraftBukkit end
                 level.setBlock(pos, blockState, 3);
                 level.scheduleTick(pos, newLiquid.getType(), spreadDelay);
             }
@@ -476,9 +_,26 @@
         public BlockState getBlockState(BlockPos pos) {
             return this.getBlockState(pos, this.getCacheKey(pos));
         }
+        // Paper start - Prevent chunk loading from fluid flowing
+        public @javax.annotation.Nullable BlockState getBlockStateIfLoaded(BlockPos pos) {
+            return this.getBlockState(pos, this.getCacheKey(pos), false);
+        }
+        // Paper end - Prevent chunk loading from fluid flowing
 
         private BlockState getBlockState(BlockPos pos, short cacheKey) {
-            return this.stateCache.computeIfAbsent(cacheKey, s -> this.level.getBlockState(pos));
+        // Paper start - Prevent chunk loading from fluid flowing
+            return getBlockState(pos, cacheKey, true);
+        }
+        private @javax.annotation.Nullable BlockState getBlockState(BlockPos pos, short packed, boolean load) {
+            BlockState blockState = this.stateCache.get(packed);
+            if (blockState == null) {
+                blockState = load ? level.getBlockState(pos) : level.getBlockStateIfLoaded(pos);
+                if (blockState != null) {
+                    this.stateCache.put(packed, blockState);
+                }
+            }
+            return blockState;
+        // Paper end - Prevent chunk loading from fluid flowing
         }
 
         public boolean isHole(BlockPos pos) {
